Netty整合WebSocket

@Slf4j
@AllArgsConstructor
public class WebSocketServer {
    private int port;

    public void run() throws InterruptedException {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss,worker)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childOption(ChannelOption.TCP_NODELAY,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //web基于http协议的解码器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //对大数据流的支持
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            //对http message进行聚合，聚合成FullHttpRequest或FullHttpResponse
                            ch.pipeline().addLast(new HttpObjectAggregator(1024 * 64));
                            //websocket服务器处理对协议，用于指定给客户端连接访问的路径
                            //该handler会帮你处理一些繁重的复杂的事
                            //会帮你处理握手动作：handshaking(close,ping,pong) ping + pong = 心跳
                            //对于websocket来讲，都是以frames进行传输的，不同的数据类型对应的frames也不同
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));
                            //添加我们的自定义channel处理器
                            ch.pipeline().addLast(new WebSocketHandler());
                        }
                    });
            log.info("服务器启动中");
            ChannelFuture future = bootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        } finally {
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }

    }
}

/**
 * TextWebSocketFrame: 在netty中，用于为websocket专门处理文本的对象，frame是消息的载体
 */
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    //用于记录和管理所有客户端的channel
    private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //获取客户端传输过来的消息
        String content = msg.text();
        log.info("接收到的数据" + content);
        clients.stream().forEach(channel -> channel.writeAndFlush(
                new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() + "接收到消息:" + content)
        ));
        //下面的方法与上面一致
//        clients.writeAndFlush(new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() +
//                "接收到消息:" + content));
    }

    /**
     * 当客户端连接服务端之后(打开连接)
     * 获取客户端的channel,并且放到ChannelGroup中去进行管理
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        clients.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        //当触发handlerRemoved,ChannelGroup会自动移除对应的客户端的channel
        //所以下面这条语句可不写
//        clients.remove(ctx.channel());
        log.info("客户端断开，channel对应的长id为:" + ctx.channel().id().asLongText());
        log.info("客户端断开，channel对应的短id为:" + ctx.channel().id().asShortText());
    }
}