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

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

maven依赖

<dependency>
    <groupId>io.nettygroupId>
    <artifactId>netty-allartifactId>
    <version>4.1.97.Finalversion>
dependency>

服务端

服务端心跳事件处理 ServerHeartbeatHandler

package com.sux.demo.websocket2;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
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) { // 写空闲

            } else if (event.state() == IdleState.ALL_IDLE) { // 读写空闲

            }
        }
    }
}

定时发送心跳的类 HeartbeatThread

package com.sux.demo.websocket2;

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

public class HeartbeatThread extends Thread {
    private ChannelGroup channelGroup;
    private boolean running = true;

    public HeartbeatThread(Channel channel) {
        channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        channelGroup.add(channel);
    }

    public HeartbeatThread(ChannelGroup channelGroup) {
        this.channelGroup = channelGroup;
    }

    @Override
    public void run() {
        while (running) {
            try {
                if (channelGroup.size() > 0) {
                    System.out.println("发送心跳");
                    for (Channel channel : channelGroup) {
                        channel.writeAndFlush(new PingWebSocketFrame());
                    }
                }

                Thread.sleep(2000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void close() {
        running = false;
    }
}

服务端封装 WebSocketServer

package com.sux.demo.websocket2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
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.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;

import java.util.concurrent.TimeUnit;

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

    private WebSocketServerHandler webSocketServerHandler;

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

        webSocketServerHandler = new WebSocketServerHandler();
    }

    public void start(int port, 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() {
                        @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(webSocketServerHandler);
                        }
                    });//给workerGroup的EventLoop对应的管道设置处理器
            //绑定端口号，启动服务端
            ChannelFuture channelFuture = bootstrap.bind(port).sync();
            System.out.println(name + " 已启动");
            // 定时发送心跳
            startHeartbeat();
            //对通道关闭进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

        }
    }

    public void startHeartbeat() {
        HeartbeatThread thread = new HeartbeatThread(webSocketServerHandler.getChannelGroup());
        thread.start();
    }

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

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

}

说明：new IdleStateHandler(5, 2, 0, TimeUnit.SECONDS)第一个参数5表示读空闲时间间隔是5秒，第二个参数2表示写空闲的时间间隔是2秒，第3个参数0表示不检测读写空闲。在ServerHeartbeatHandler代码中，读写空闲时断开与客户端的连接。

服务端消息处理 WebSocketServerHandler

package com.sux.demo.websocket2;

import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.concurrent.GlobalEventExecutor;

@ChannelHandler.Sharable
public class WebSocketServerHandler extends SimpleChannelInboundHandler