Netty实现WebSocket长连接

约定不等于承诺〃 2022-09-06 13:27 213阅读 0赞

WebSocket协议是使用单个TCP连接进行全双工通信的协议，浏览器与服务器使用一个长连接，可以双向数据传输，且服务器可以主动给浏览器推送数据。Netty是通过给通道绑定的管道，添加一个WebSocketServerProtocleHandler处理器，将http协议升级为WebSocket协议的，数据传输采用帧WebSocketFrame的形式，比如常用的文本帧TextWebSocketFrame传输文本数据。

服务器

package com.tech.netty.netty.websocket;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    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.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.stream.ChunkedWriteHandler;
    
    /**
     * @author lw
     * @since 2021/8/20
     */
    public class MyServer {
        public static void main(String[] args) throws InterruptedException {
            //创建2个线程组
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                //因为基于http协议 使用http的编码器和解码器
                                pipeline.addLast("HttpServerCodec",new HttpServerCodec())
                                        //以块方式写 添加ChunkedWriteHandler处理器
                                        .addLast("ChunkedWriteHandler", new ChunkedWriteHandler())
                                        //http传输过程中会分段 HttpObjectAggregator可以将多个段聚合
                                        //这就是为什么，当浏览器发送大量数据时，就会发多次http请求
                                        .addLast("HttpObjectAggregator",new HttpObjectAggregator(8192))
                                        //对应websocket 它的数据是以帧(WebSocketFrame)形式传递
                                        //浏览器请求时ws://localhost:7000/hello表示请求的uri
                                        //WebSocketServerProtocolHandler的核心功能是将http协议升级为ws协议 并保持长连接
                                        //是通过一个状态码 101
                                        .addLast("WebSocketServerProtocolHandler",new WebSocketServerProtocolHandler("/hello"))
                                        //自定义的handler做业务逻辑处理
                                        .addLast(new MyTextWebSocketFrameHandler());
                            }
                        });
                //启动服务器
                ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
                channelFuture.channel().closeFuture().sync();
            }finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

package com.tech.netty.netty.websocket;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    
    import java.time.LocalDateTime;
    
    /**
     * @author lw
     * @since 2021/8/20
     */
    //TextWebSocketFrame表示一个文本帧
    public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
            System.out.println("服务端接收到消息"+msg.text());
            //回复消息
            ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间 "+ LocalDateTime.now()+" "+msg.text()));
        }
    
        //当web客户端连接后触发该方法
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            System.out.println("handlerAdded 被调用 "+ctx.channel().id().asLongText());
            System.out.println("handlerAdded 被调用 "+ctx.channel().id().asShortText());
        }
    
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            //id表示唯一 longText是唯一的 shortText不是惟一的
            System.out.println("handlerRemoved 被调用 "+ctx.channel().id().asLongText());
            System.out.println("handlerRemoved 被调用 "+ctx.channel().id().asShortText());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("异常发生 "+cause.getMessage());
            //关闭连接
            ctx.close();
        }
    }

客户端

<!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>Title</title>
    </head>
    <body>
    <script>
        var socket;
        //判断当前浏览器是否支持websocket
        if(window.WebSocket) {
            //go on
            socket = new WebSocket("ws://localhost:7000/hello");
            //相当于channelReado, ev 收到服务器端回送的消息
            socket.onmessage = function (ev) {
                var rt = document.getElementById("responseText");
                rt.value = rt.value + "\n" + ev.data;
            }
    
            //相当于连接开启(感知到连接开启)
            socket.onopen = function (ev) {
                var rt = document.getElementById("responseText");
                rt.value = "连接开启了.."
            }
    
            //相当于连接关闭(感知到连接关闭)
            socket.onclose = function (ev) {
    
                var rt = document.getElementById("responseText");
                rt.value = rt.value + "\n" + "连接关闭了.."
            }
        } else {
            alert("当前浏览器不支持websocket")
        }
    
        //发送消息到服务器
        function send(message) {
            if(!window.socket) { //先判断socket是否创建好
                return;
            }
            if(socket.readyState == WebSocket.OPEN) {
                //通过socket 发送消息
                socket.send(message)
            } else {
                alert("连接没有开启");
            }
        }
    </script>
        <form onsubmit="return false">
            <textarea name="message" style="height: 300px; width: 300px"></textarea>
            <input type="button" value="发生消息" onclick="send(this.form.message.value)">
            <textarea id="responseText" style="height: 300px; width: 300px"></textarea>
            <input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''">
        </form>
    </body>
    </html>

程序运行后 测试效果如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70]: /images/20220829/967946e7de5c4acaa071bb6c4233b243.png