java netty 实现 websocket 服务端和客户端双向通信 实现心跳和断线重连 完整示例
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", null, false, 65536, false, false, false, 10000)); 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 PongWebSocketFrame) { System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发来的心跳：PONG"); } 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(); } 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; 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(); } } } 客户端
客户端心跳 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.codec.http.websocketx.PongWebSocketFrame; 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 PongWebSocketFrame()); // 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; 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(); } } } 测试
测试一
步骤：先启动服务端，再启动客户端
现象：客户端与服务端互发消息，消息发完后，互发心跳
测试二
步骤：先启动服务端，再启动客户端，然后关闭服务端，过一会再启动服务端
现象：客户端断线重连，通信恢复，正常发消息和心跳
测试三
步骤：先启动客户端，过一会再启动服务端
现象：服务端启动后，客户端连上服务端，正常通信，互发消息，消息发完互发心跳
遇到的问题
以上测试，客户端可以收到服务端发送的PONG心跳，但是服务端无法收到客户端发送的PING心跳，却能收到客户端发送的PONG心跳
