java netty 实现 websocket 服务端和客户端双向通信实现心跳和断线重连完整示例

标签：netty websocket 示例 ctx io import public channel

maven依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.97.Final</version>
</dependency>

服务端

一个接口 IGetHandshakeFuture

package com.sux.demo.websocket2;

import io.netty.channel.ChannelPromise;

public interface IGetHandshakeFuture {
    ChannelPromise getHandshakeFuture();
}

服务端心跳 ServerHeartbeatHandler

package com.sux.demo.websocket2;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) { // 读空闲
                // System.out.println("关闭客户端连接, channel id=" + ctx.channel().id());
                // ctx.channel().close();
            } else if (event.state() == IdleState.WRITER_IDLE) { // 写空闲
                System.out.println("服务端向客户端发送心跳");
                ctx.writeAndFlush(new PongWebSocketFrame());
            } else if (event.state() == IdleState.ALL_IDLE) { // 读写空闲

            }
        }
    }
}

服务端封装 WebSocketServer

package com.sux.demo.websocket2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

public class WebSocketServer {
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    public WebSocketServer() {
        //创建两个线程组 boosGroup、workerGroup
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
    }

    public void start(int port, WebSocketServerHandler handler, String name) {
        try {
            //创建服务端的启动对象，设置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //设置两个线程组boosGroup和workerGroup
            bootstrap.group(bossGroup, workerGroup)
                    //设置服务端通道实现类型
                    .channel(NioServerSocketChannel.class)
                    //设置线程队列得到连接个数
                    .option(ChannelOption.SO_BACKLOG, 128)
                    //设置保持活动连接状态
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //使用匿名内部类的形式初始化通道对象
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //给pipeline管道设置处理器
                            socketChannel.pipeline().addLast(new HttpServerCodec());
                            socketChannel.pipeline().addLast(new HttpObjectAggregator(65536));
                            socketChannel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
                            socketChannel.pipeline().addLast(new IdleStateHandler(5, 2, 0, TimeUnit.SECONDS));
                            socketChannel.pipeline().addLast(new ServerHeartbeatHandler());
                            socketChannel.pipeline().addLast(handler);
                        }
                    });//给workerGroup的EventLoop对应的管道设置处理器
            //绑定端口号，启动服务端
            ChannelFuture channelFuture = bootstrap.bind(port).sync();
            System.out.println(name + " 已启动");
            //对通道关闭进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

        }
    }
}

服务端消息处理 WebSocketServerHandler

package com.sux.demo.websocket2;

import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;

import java.util.ArrayList;
import java.util.List;

@ChannelHandler.Sharable
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private List<Channel> channelList;

    public WebSocketServerHandler() {
        channelList = new ArrayList<>();
    }

    public boolean hasClient() {
        return channelList.size() > 0;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof PingWebSocketFrame) {
            System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发来的心跳：PING");
        }

        if (msg instanceof TextWebSocketFrame) {
            TextWebSocketFrame frame = (TextWebSocketFrame) msg;
            System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发来的消息：" + frame.text());
            /*for (Channel channel : channelList) {
                if (!ctx.channel().id().toString().equals(channel.id().toString())) {
                    channel.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer(frame.text(), CharsetUtil.UTF_8)));
                    System.out.println("服务端向客户端 " + channel.id().toString() + " 转发消息：" + frame.text());
                }
            }*/
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        channelList.add(ctx.channel());
        System.out.println("客户端连接：" + ctx.channel().id().toString());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        channelList.remove(ctx.channel());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        channelList.remove(ctx.channel());
    }

    public void send(String text) {
        for (Channel channel : channelList) {
            channel.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer(text, CharsetUtil.UTF_8)));
        }
    }
}

服务端测试主机 WebSocketClientHost

package com.sux.demo.websocket2;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.CharsetUtil;

public class WebSocketClientHost {
    public static void main(String[] args) {
        WebSocketClient webSocketClient = new WebSocketClient();

        SendDataToServerThread thread = new SendDataToServerThread(webSocketClient);
        thread.start();

        webSocketClient.connect("127.0.0.1", 40005, "WebSocket客户端");
    }
}

class SendDataToServerThread extends Thread {
    private WebSocketClient webSocketClient;

    private int index = 1;

    public SendDataToServerThread(WebSocketClient webSocketClient) {
        this.webSocketClient = webSocketClient;
    }

    @Override
    public void run() {
        try {
            while (index <= 5) {
                Channel channel = webSocketClient.getChannel();
                if (channel != null && channel.isActive()) {
                    String msg = "客户端发送的测试消息, index= " + index;
                    channel.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8)));
                    index++;
                }
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

客户端

客户端心跳 ClientHeartbeatHandler

package com.sux.demo.websocket2;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;


public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) { // 读空闲
                System.out.println("断线重连");
                ctx.channel().close();
            } else if (event.state() == IdleState.WRITER_IDLE) { // 写空闲
                System.out.println("客户端向服务端发送心跳");
                ctx.writeAndFlush(new PingWebSocketFrame());
                // ctx.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer("PING", CharsetUtil.UTF_8)));
            } else if (event.state() == IdleState.ALL_IDLE) { // 读写空闲

            }
        }
    }
}

客户端封装 WebSocketClient

package com.sux.demo.websocket2;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.timeout.IdleStateHandler;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.TimeUnit;

public class WebSocketClient {
    private NioEventLoopGroup eventExecutors;

    private Channel channel;

    public WebSocketClient() {
        eventExecutors = new NioEventLoopGroup();
    }

    public Channel getChannel() {
        return channel;
    }

    public void connect(String ip, int port, String name) {
        try {
            WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(
                    new URI("ws://" + ip + ":" + port + "/websocket"), WebSocketVersion.V13, null, false, new DefaultHttpHeaders());
            WebSocketClientHandler handler = new WebSocketClientHandler(handshaker);
            ClientHeartbeatHandler heartbeatHandler = new ClientHeartbeatHandler();

            //创建bootstrap对象，配置参数
            Bootstrap bootstrap = new Bootstrap();
            //设置线程组
            bootstrap.group(eventExecutors)
                    //设置客户端的通道实现类型
                    .channel(NioSocketChannel.class)
                    //使用匿名内部类初始化通道
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //添加客户端通道的处理器
                            ch.pipeline().addLast(new HttpClientCodec());
                            ch.pipeline().addLast(new HttpObjectAggregator(65536));
                            ch.pipeline().addLast(new WebSocketClientProtocolHandler(handshaker, true, false));
                            ch.pipeline().addLast(new IdleStateHandler(5, 2, 0, TimeUnit.SECONDS));
                            ch.pipeline().addLast(heartbeatHandler);
                            ch.pipeline().addLast(handler);
                        }
                    });

            // 连接服务端
            ChannelFuture channelFuture = bootstrap.connect(ip, port);

            // 在连接关闭后尝试重连
            channelFuture.channel().closeFuture().addListener(future -> {
                try {
                    Thread.sleep(2000);
                    System.out.println("重新连接");
                    connect(ip, port, name); // 重新连接
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });

            channelFuture.sync();

            // 等待握手完成
            // IGetHandshakeFuture getHadnshakeFuture = handler;
            // getHadnshakeFuture.getHandshakeFuture().sync();

            channel = channelFuture.channel();
            System.out.println(name + " 已启动");

            //对通道关闭进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException | URISyntaxException e) {
            e.printStackTrace();
        } finally {

        }
    }

}

客户端消息处理 WebSocketClientHandler

package com.sux.demo.websocket2;

import io.netty.channel.*;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.websocketx.*;

@ChannelHandler.Sharable
public class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> implements IGetHandshakeFuture {
    private WebSocketClientHandshaker handshaker;
    private ChannelPromise handshakeFuture;

    public WebSocketClientHandler(WebSocketClientHandshaker handshaker) {
        this.handshaker = handshaker;
    }

    public ChannelPromise getHandshakeFuture() {
        return this.handshakeFuture;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (!handshaker.isHandshakeComplete()) {
            try {
                handshaker.finishHandshake(ctx.channel(), (FullHttpResponse) msg);
                handshakeFuture.setSuccess();
            } catch (WebSocketHandshakeException e) {
                handshakeFuture.setFailure(e);
            }
            return;
        }

        if (msg instanceof PongWebSocketFrame) {
            System.out.println("收到服务端" + ctx.channel().remoteAddress() + "发来的心跳：PONG");
        }

        if (msg instanceof TextWebSocketFrame) {
            TextWebSocketFrame frame = (TextWebSocketFrame) msg;
            System.out.println("收到服务端" + ctx.channel().remoteAddress() + "发来的消息：" + frame.text()); // 接收服务端发送过来的消息
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        handshakeFuture = ctx.newPromise();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // handshaker.handshake(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

客户端测试主机 WebSocketServerHost

package com.sux.demo.websocket2;

public class WebSocketServerHost {
    public static void main(String[] args) {
        WebSocketServerHandler handler = new WebSocketServerHandler();
        WebSocketServer webSocketServer = new WebSocketServer();

        SendDataToClientThread thread = new SendDataToClientThread(handler);
        thread.start();

        webSocketServer.start(40005, handler, "WebSocket服务端");
    }
}

class SendDataToClientThread extends Thread {
    private WebSocketServerHandler handler;

    private int index = 1;

    public SendDataToClientThread(WebSocketServerHandler handler) {
        this.handler = handler;
    }

    @Override
    public void run() {
        try {
            while (index <= 5) {
                if (handler.hasClient()) {
                    String msg = "服务端发送的测试消息, index= " + index;
                    handler.send(msg);
                    index++;
                }
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

测试

测试一

步骤：先启动服务端，再启动客户端
现象：客户端与服务端互发消息，消息发完后，互发心跳

测试二

步骤：先启动服务端，再启动客户端，然后关闭服务端，过一会再启动服务端
现象：客户端断线重连，通信恢复，正常发消息和心跳

测试三

步骤：先启动客户端，过一会再启动服务端
现象：服务端启动后，客户端连上服务端，正常通信，互发消息，消息发完互发心跳

遇到的问题

以上测试，客户端可以收到服务端发送的心跳，但是服务端无法收到客户端发送的心跳，导致服务端触发读空闲，从而关闭连接(代码已注释掉)

标签：netty,websocket,示例,ctx,io,import,public,channel
From： https://www.cnblogs.com/s0611163/p/18194177

java netty 实现 websocket 服务端和客户端双向通信实现心跳和断线重连完整示例

maven依赖

服务端

一个接口 IGetHandshakeFuture

服务端心跳 ServerHeartbeatHandler

服务端封装 WebSocketServer

服务端消息处理 WebSocketServerHandler

服务端测试主机 WebSocketClientHost

客户端

客户端心跳 ClientHeartbeatHandler

客户端封装 WebSocketClient

客户端消息处理 WebSocketClientHandler

客户端测试主机 WebSocketServerHost

测试

测试一

测试二

测试三

遇到的问题

相关文章

赞助商

阅读排行

java netty 实现 websocket 服务端和客户端双向通信 实现心跳和断线重连 完整示例

maven依赖

服务端

一个接口 IGetHandshakeFuture

服务端心跳 ServerHeartbeatHandler

服务端封装 WebSocketServer

服务端消息处理 WebSocketServerHandler

服务端测试主机 WebSocketClientHost

客户端

客户端心跳 ClientHeartbeatHandler

客户端封装 WebSocketClient

客户端消息处理 WebSocketClientHandler

客户端测试主机 WebSocketServerHost

测试

测试一

测试二

测试三

遇到的问题

相关文章

赞助商

阅读排行

java netty 实现 websocket 服务端和客户端双向通信实现心跳和断线重连完整示例