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

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

1. <dependency>
2.     <groupId>io.netty</groupId>
3.     <artifactId>netty-all</artifactId>
4.     <version>4.1.97.Final</version>
5. </dependency>

复制代码

服务端

一个接口 IGetHandshakeFuture

1. package com.sux.demo.websocket2;
3. import io.netty.channel.ChannelPromise;
5. public interface IGetHandshakeFuture {
6.     ChannelPromise getHandshakeFuture();
7. }

复制代码

服务端心跳 ServerHeartbeatHandler

1. package com.sux.demo.websocket2;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.channel.ChannelInboundHandlerAdapter;
5. import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
6. import io.netty.handler.timeout.IdleState;
7. import io.netty.handler.timeout.IdleStateEvent;
9. public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {
10.     @Override
11.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
12.         if (evt instanceof IdleStateEvent) {
13.             IdleStateEvent event = (IdleStateEvent) evt;
14.             if (event.state() == IdleState.READER_IDLE) { // 读空闲
15.                 System.out.println("关闭客户端连接, channel id=" + ctx.channel().id());
16.                 ctx.channel().close();
17.             } else if (event.state() == IdleState.WRITER_IDLE) { // 写空闲
18.                 System.out.println("服务端向客户端发送心跳");
19.                 ctx.writeAndFlush(new PongWebSocketFrame());
20.             } else if (event.state() == IdleState.ALL_IDLE) { // 读写空闲
22.             }
23.         }
24.     }
25. }

复制代码

服务端封装 WebSocketServer

1. package com.sux.demo.websocket2;
3. import io.netty.bootstrap.ServerBootstrap;
4. import io.netty.channel.*;
5. import io.netty.channel.nio.NioEventLoopGroup;
6. import io.netty.channel.socket.SocketChannel;
7. import io.netty.channel.socket.nio.NioServerSocketChannel;
8. import io.netty.handler.codec.http.HttpObjectAggregator;
9. import io.netty.handler.codec.http.HttpServerCodec;
10. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
11. import io.netty.handler.timeout.IdleStateHandler;
13. import java.util.concurrent.TimeUnit;
15. public class WebSocketServer {
16.     private EventLoopGroup bossGroup;
17.     private EventLoopGroup workerGroup;
19.     public WebSocketServer() {
20.         //创建两个线程组 boosGroup、workerGroup
21.         bossGroup = new NioEventLoopGroup();
22.         workerGroup = new NioEventLoopGroup();
23.     }
25.     public void start(int port, WebSocketServerHandler handler, String name) {
26.         try {
27.             //创建服务端的启动对象，设置参数
28.             ServerBootstrap bootstrap = new ServerBootstrap();
29.             //设置两个线程组boosGroup和workerGroup
30.             bootstrap.group(bossGroup, workerGroup)
31.                     //设置服务端通道实现类型
32.                     .channel(NioServerSocketChannel.class)
33.                     //设置线程队列得到连接个数
34.                     .option(ChannelOption.SO_BACKLOG, 128)
35.                     //设置保持活动连接状态
36.                     .childOption(ChannelOption.SO_KEEPALIVE, true)
37.                     //使用匿名内部类的形式初始化通道对象
38.                     .childHandler(new ChannelInitializer<SocketChannel>() {
39.                         @Override
40.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
41.                             //给pipeline管道设置处理器
42.                             socketChannel.pipeline().addLast(new HttpServerCodec());
43.                             socketChannel.pipeline().addLast(new HttpObjectAggregator(65536));
44.                             socketChannel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket", null, false, 65536, false, false, false, 10000));
45.                             socketChannel.pipeline().addLast(new IdleStateHandler(5, 2, 0, TimeUnit.SECONDS));
46.                             socketChannel.pipeline().addLast(new ServerHeartbeatHandler());
47.                             socketChannel.pipeline().addLast(handler);
48.                         }
49.                     });//给workerGroup的EventLoop对应的管道设置处理器
50.             //绑定端口号，启动服务端
51.             ChannelFuture channelFuture = bootstrap.bind(port).sync();
52.             System.out.println(name + " 已启动");
53.             //对通道关闭进行监听
54.             channelFuture.channel().closeFuture().sync();
55.         } catch (InterruptedException e) {
56.             e.printStackTrace();
57.         } finally {
59.         }
60.     }
61. }

复制代码

服务端消息处理惩罚 WebSocketServerHandler

1. package com.sux.demo.websocket2;
3. import io.netty.buffer.Unpooled;
4. import io.netty.channel.*;
5. import io.netty.handler.codec.http.websocketx.*;
6. import io.netty.util.CharsetUtil;
8. import java.util.ArrayList;
9. import java.util.List;
11. @ChannelHandler.Sharable
12. public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
14.     private List<Channel> channelList;
16.     public WebSocketServerHandler() {
17.         channelList = new ArrayList<>();
18.     }
20.     public boolean hasClient() {
21.         return channelList.size() > 0;
22.     }
24.     @Override
25.     protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
26.         if (msg instanceof PingWebSocketFrame) {
27.             System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发来的心跳：PING");
28.         }
30.         if (msg instanceof PongWebSocketFrame) {
31.             System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发来的心跳：PONG");
32.         }
34.         if (msg instanceof TextWebSocketFrame) {
35.             TextWebSocketFrame frame = (TextWebSocketFrame) msg;
36.             System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发来的消息：" + frame.text());
37.             /*for (Channel channel : channelList) {
38.                 if (!ctx.channel().id().toString().equals(channel.id().toString())) {
39.                     channel.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer(frame.text(), CharsetUtil.UTF_8)));
40.                     System.out.println("服务端向客户端 " + channel.id().toString() + " 转发消息：" + frame.text());
41.                 }
42.             }*/
43.         }
44.     }
46.     @Override
47.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
48.         channelList.add(ctx.channel());
49.         System.out.println("客户端连接：" + ctx.channel().id().toString());
50.     }
52.     @Override
53.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
54.         channelList.remove(ctx.channel());
55.     }
57.     @Override
58.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
59.         ctx.flush();
60.     }
62.     @Override
63.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
64.         ctx.close();
65.     }
67.     public void send(String text) {
68.         for (Channel channel : channelList) {
69.             channel.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer(text, CharsetUtil.UTF_8)));
70.         }
71.     }
72. }

复制代码

服务端测试主机 WebSocketClientHost

1. package com.sux.demo.websocket2;
3. public class WebSocketServerHost {
4.     public static void main(String[] args) {
5.         WebSocketServerHandler handler = new WebSocketServerHandler();
6.         WebSocketServer webSocketServer = new WebSocketServer();
8.         SendDataToClientThread thread = new SendDataToClientThread(handler);
9.         thread.start();
11.         webSocketServer.start(40005, handler, "WebSocket服务端");
12.     }
13. }
15. class SendDataToClientThread extends Thread {
16.     private WebSocketServerHandler handler;
18.     private int index = 1;
20.     public SendDataToClientThread(WebSocketServerHandler handler) {
21.         this.handler = handler;
22.     }
24.     @Override
25.     public void run() {
26.         try {
27.             while (index <= 5) {
28.                 if (handler.hasClient()) {
29.                     String msg = "服务端发送的测试消息, index = " + index;
30.                     handler.send(msg);
31.                     index++;
32.                 }
33.                 Thread.sleep(1000);
34.             }
35.         } catch (Exception e) {
36.             e.printStackTrace();
37.         }
38.     }
39. }

复制代码

客户端消息处理惩罚 WebSocketClientHandler

1. package com.sux.demo.websocket2;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.channel.ChannelInboundHandlerAdapter;
5. import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
6. import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
7. import io.netty.handler.timeout.IdleState;
8. import io.netty.handler.timeout.IdleStateEvent;
11. public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {
12.     @Override
13.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
14.         if (evt instanceof IdleStateEvent) {
15.             IdleStateEvent event = (IdleStateEvent) evt;
16.             if (event.state() == IdleState.READER_IDLE) { // 读空闲
17.                 System.out.println("断线重连");
18.                 ctx.channel().close();
19.             } else if (event.state() == IdleState.WRITER_IDLE) { // 写空闲
20.                 System.out.println("客户端向服务端发送心跳");
21.                 ctx.writeAndFlush(new PongWebSocketFrame());
22.                 // ctx.writeAndFlush(new PingWebSocketFrame());
23.                 // ctx.writeAndFlush(new TextWebSocketFrame(Unpooled.copiedBuffer("PING", CharsetUtil.UTF_8)));
24.             } else if (event.state() == IdleState.ALL_IDLE) { // 读写空闲
26.             }
27.         }
28.     }
29. }

复制代码

客户端测试主机 WebSocketServerHost

[code]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