Netty协议开发（WebSocket）

﹏ヽ暗。殇╰゛Y 2022-05-14 15:26 317阅读 0赞

WebSocket是H5开始提供的一种游览器与服务器间进行双全工通信的网络技术，WebSocket通信协议与2011年被IETF定义为标准RFC6455，WebSocket API被W3c定义为标准。

其特点如下：

1.  单一的TCP连接，采用双全工模式通信；
2.  对代理、防火墙和路由器透明；
3.  无头部信息、Cookie和身份验证；
4.  无安全开销；
5.  通过“ping/pong”帧保持链路激活；
6.  服务器可以主动传递消息给客户端，不再需要客户端轮询。

### WebSocket连接建立： ###

建立一个WebSocket连接客户端游览器首先要向服务端发起一个HTTP请求，这个请求和通常HTTP不同，包含了一些附加头信息，其中附加头信息“Upgrade:WebSocket”表明这是一个申请协议升级的HTTP请求。服务端解析这些附加的头信息，然后生成应答信息返回给客户端，客户端和服务端的WebSocket连接建立。双方可以通过这个连接通道自由的传递信息，知道客户端或者服务端某一方主动关闭连接。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70][]

Sec-WebSocket-Key是随机的，服务器端会用这些数据来构造一个SHA-1的信息摘要，把Sec-WebSocket-Key加上一个魔幻字符串，再使用SHA-1加密，然后进行BASE-64编码，将结果作为Sec-WebSocket-Accept头的值，返回给客户端。

### WebSocket生命周期： ###

握手成功后，服务端和客户端就可以通过“messages”的方式进行通信，一个消息由一个或多个帧组成，WebSocket的消息并不一定对应一个特定网络层的帧，它可以被分割成为多个帧或者被合并。

帧都有自己对应的类型，属于同一个消息的多个帧具有相同类型的数据。从广义上讲，数据类型可以是文本数据、二进制数据和控制帧（协议级信息，如信息）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70 1][]

### WebSocket连接关闭 ###

为关闭WebSocket连接，客户端和服务端需要通过一个安全的方法关闭底层TCP连接以及TLS会话。如果合适，丢弃任何可能已经接受的字节，必要时可以通过任何可用手段关闭连接。

底层的TCP连接，在正常情况下，首先应该由服务器关闭。异常情况下客户端可以发起TCP Close。因此当服务器被指示关闭WebSocket连接时，它应该立即发起一个TCP Close操作；客户端应该等待服务器的TCP Close。

WebSocket的握手关闭消息带有一个状态码和一个可选的关闭原因，它必须按照协议要求发送一个Close控制帧，当对端接收到关闭控制帧指令时，需要主动关闭WebSocket连接。

ok~下面看下服务端代码：

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    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.stream.ChunkedWriteHandler;
    
    public class WebSocketServer {
    
        public void run() {
            // 服务端启动辅助类，用于设置TCP相关参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 获取Reactor线程池
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();
            // 设置为主从线程模型
            bootstrap.group(bossGroup, workGroup)
                    // 设置服务端NIO通信类型
                    .channel(NioServerSocketChannel.class)
                    // 设置ChannelPipeline，也就是业务职责链，由处理的Handler串联而成，由从线程池处理
                    .childHandler(new ChannelInitializer<Channel>() {
                        // 添加处理的Handler，通常包括消息编解码、业务处理，也可以是日志、权限、过滤等
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            // 获取职责链
                            ChannelPipeline pipeline = ch.pipeline();
                            //将请求和应答消息编码或者解码消息为HTTP消息。
                            pipeline.addLast("http-codec", new HttpServerCodec());
                            //它的目的是将HTTP消息的多个部分组合成一条完整的HTTP消息
                            pipeline.addLast("aggregator", new HttpObjectAggregator(65535));
                            //用于向客户端发送HTML5文件，用于支持游览器和服务端进行WebSocket通信
                            pipeline.addLast("http-chunked", new ChunkedWriteHandler());
                            pipeline.addLast("handler", new WebSocketHandler());
                        }
                    })
                    // bootstrap 还可以设置TCP参数，根据需要可以分别设置主线程池和从线程池参数，来优化服务端性能。
                    // 其中主线程池使用option方法来设置，从线程池使用childOption方法设置。
                    // backlog表示主线程池中在套接口排队的最大数量，队列由未连接队列（三次握手未完成的）和已连接队列
                    .option(ChannelOption.SO_BACKLOG, 5)
                    // 表示连接保活，相当于心跳机制，默认为7200s
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
    
            try {
                // 绑定端口，启动select线程，轮询监听channel事件，监听到事件之后就会交给从线程池处理
                Channel channel = bootstrap.bind(8082).sync().channel();
                System.out.println("WebSocket服务器开启，port:8081");
                // 等待服务端口关闭
                channel.closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 优雅退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) {
            new WebSocketServer().run();
        }
    
    }

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.handler.codec.http.*;
    import io.netty.handler.codec.http.websocketx.*;
    import io.netty.util.CharsetUtil;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    public class WebSocketHandler extends ChannelInboundHandlerAdapter {
        //用于websocket握手的处理类
        private WebSocketServerHandshaker handshaker;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            if (msg instanceof FullHttpRequest) {
                // websocket连接请求
                handleHttpRequest(ctx, (FullHttpRequest)msg);
            } else if (msg instanceof WebSocketFrame) {
                // websocket业务处理
                handleWebSocketRequest(ctx, (WebSocketFrame)msg);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            ctx.close();
        }
    
        private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
            // Http解码失败，向服务器指定传输的协议为Upgrade：websocket
            if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
                sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
                return;
            }
            // 握手相应处理,创建websocket握手的工厂类，本机测试
            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:8082/zzf", null, false);
            // 根据工厂类和HTTP请求创建握手类
            handshaker = wsFactory.newHandshaker(req);
            if (handshaker == null) {
                // 不支持websocket
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                // 通过它构造握手响应消息返回给客户端
                handshaker.handshake(ctx.channel(), req);
            }
        }
    
        private void handleWebSocketRequest(ChannelHandlerContext ctx, WebSocketFrame req) throws Exception {
    
            //判断是否是关闭链路指令
            if (req instanceof CloseWebSocketFrame) {
                // 关闭websocket连接
                handshaker.close(ctx.channel(), (CloseWebSocketFrame)req.retain());
                return;
            }
            //判断是否是Ping消息
            if (req instanceof PingWebSocketFrame) {
                ctx.channel().write(new PongWebSocketFrame(req.content().retain()));
                return;
            }
            //本例支持文本消息，不支持二进制消息
            if (!(req instanceof TextWebSocketFrame)) {
                throw new UnsupportedOperationException("当前只支持文本消息，不支持二进制消息");
            }
            if (ctx == null || this.handshaker == null || ctx.isRemoved()) {
                throw new Exception("尚未握手成功，无法向客户端发送WebSocket消息");
            }
            //输入的信息
            System.out.println(((TextWebSocketFrame) req).text());
            //返回应答消息：
            ctx.channel().writeAndFlush(new TextWebSocketFrame(((TextWebSocketFrame)req).text()+
                    "    当前时间:   "+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
        }
    
        private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
            // BAD_REQUEST(400) 客户端请求错误返回的应答消息
            if (res.status().code() != 200) {
                // 将返回的状态码放入缓存中，Unpooled没有使用缓存池
                ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
                res.content().writeBytes(buf);
                buf.release();
                HttpUtil.setContentLength(res, res.content().readableBytes());
            }
            // 发送应答消息
            ChannelFuture cf = ctx.channel().writeAndFlush(res);
            // 非法连接直接关闭连接
            if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
                cf.addListener(ChannelFutureListener.CLOSE);
            }
        }
    
    }

客户端HTML代码：

<!DOCTYPE html>
    <html>
    <head>
    <meta charset="UTF-8">
    <title>WebSocket Chat</title>
    </head>
    <body>
        <script type="text/javascript">
            var socket;
            if (!window.WebSocket) {
                window.WebSocket = window.MozWebSocket;
            }
            if (window.WebSocket) {
                socket = new WebSocket("ws://localhost:8082/zzf");
                socket.onopen = function(event) {
                    var ta = document.getElementById('responseText');
                    ta.value = "连接开启!";
                };
                socket.onclose = function(event) {
                    var ta = document.getElementById('responseText');
                    ta.value = ta.value + "连接被关闭";
                };
                socket.onmessage = function(event) {
                    var ta = document.getElementById('responseText');
                    ta.value = ta.value + '\n' + event.data;
                };
            } else {
                alert("你的浏览器不支持 WebSocket！");
            }
     
            function send(message) {
                if (!window.WebSocket) {
                    return;
                }
                if (socket.readyState == WebSocket.OPEN) {
                    socket.send(message);
                } else {
                    alert("连接没有开启.");
                }
            }
        </script>
        <form onsubmit="return false;">
            <h3>WebSocket 聊天室：</h3>
            <textarea id="responseText" style="width: 500px; height: 300px;"></textarea>
            <br> 
            <input type="text" name="message"  style="width: 300px" value="Welcome to www.waylau.com">
            <input type="button" value="发送消息" onclick="send(this.form.message.value)">
            <input type="button" onclick="javascript:document.getElementById('responseText').value=''" value="清空聊天记录">
        </form>
        <br> 
        <br> 
    </body>
    </html>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70]: /images/20220503/c644bf23e4874c6191d52449d6c6bba3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70 1]: /images/20220503/101064b3fb4747c19c27747e48327f5d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70 2]: /images/20220503/02fb0defe745420897a1504bf9a890ee.png