【如何构建商业级别聊天系统】 Netty 构建 WebSocket 服务上篇

梦里梦外; 2022-09-12 09:52 183阅读 0赞

### 写在前面 ###

对于 WebSocket 协议感兴趣的同学可以参考 [rfc6455][]

以及我之前的文章 [WebSocket 协议][WebSocket]

本篇将带领大家构建一个最初的 WebSocket 原型，以及前端页面配合服务发送简单消息。后续的服务将基于此进行演化。

在开始之前我还是想再聊聊 WebSocket 协议，本篇从此开始。

#### 一、再聊 WebSocket ####

**不可避免的 WebSocket 是什么？**

它是一个协议，**基于 TCP 协议**且**利用 Http 协议进行握手**的一个**全双工**的协议。

相比于 Http 的单向请求，WebSocket 可以做到**双向通信**，即服务端可以向客户端推送消息。这个特性也是我们选用 WebSocket 的最重要的理由。

**WebSocket 握手是一个 Http 升级协议升级请求**

为了兼容基于 Http 的服务端和中间代理，所以 Http 客户端 和 WebSocket 客户端可以使用一个端口与 Server 通信，为此 WebSocket 客户端的握手就是一个 Http 的升级请求

GET /chat HTTP/1.1
    Host: server.example.com
    Upgrade: websocket
    Connection: Upgrade
    Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
    Origin: http://example.com
    Sec-WebSocket-Protocol: chat, superchat
    Sec-WebSocket-Version: 13

**数据传输**

WebSocket 的数据传输采用的是 **二进制帧** 的形式。

该 二进制帧 由 **帧头** 和 **帧内容** 构成

\*\*帧内容 \*\*就是我们要传输的真正的数据内容，而帧头是用于标识帧内容的。

\*\*帧头 \*\*由 FIN 结束位；RSV1、2、3 ；OpenCode 帧类型；MASK 掩码；Payload len 帧内容长度；MASK Key 掩码 key 组成。

1.  FIN 结束位：用于标识这条消息的结束，一条消息可以由多个二进制帧构成，所以需要结束位来标识该条消息的结束。
2.  RSV1、RSV2、RSV3：必须为 0 ，除非协商了拓展。
3.  OpenCode 帧类型：用于标识帧类型，即有效载荷的数据类型，4 个 bit 位，用 16 进制来表示如下 （ps %x 在 C 中表示 16 进制开头）
    
    1.      %x0 表示继续帧
            
            %x1 表示文本帧
            
            %x2 表示二进制帧
            
            %x3-7 保留用于进一步的非控制帧
            
            %x8 表示连接关闭
            
            %x9 表示 ping
            
            %xA表示 pong
            
            %xB-F 保留用于进一步的控制帧
4.  MASK 掩码 ： 客户端向服务端发送数据 MASK 必须为 1 ，而服务端向客户端发送 MASK 必须为 0。 该字段与 MASK Key 字段对应。
5.  Payload len 有效载荷长度：该字段表示帧内容长度，长度为 7 位或者 7+16位、7+64 位。前 7 位如果是 0-125 则直接表示 Payload len，如果是 126 则拓展 16 位，这 2 个字节表示 Payload len。如果前七位为 127 则拓展 64 位，这 8 个字节表示 Payload len 。所以有效载荷最大为 2^64 个字节
6.  Masking-Key 掩码密匙：如果 MASK 为 1 则存在，0 则不存在 。由 32 bit 即 4 字节组成。
7.  Payload Data： 由 Extension Data 和 Application Data 组成，Extension Data 默认为 0 除非在握手明确协商。

ok WebSocket 协议复习到这里，相信大家对该协议已经有一个感性的认识了，接下来构建 WebSocket 服务。

#### 二、构建 WebSocket 服务 ####

import com.star.im.websocket.handler.CommonConnectHandler;
    import com.star.im.websocket.handler.MqttWebSocketCodec;
    import com.star.im.websocket.handler.TextWebSocketFrameHandler;
    import com.star.im.websocket.handler.demo.DiscardServerHandler;
    import io.netty.bootstrap.ServerBootstrap;
    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.WebSocketServerProtocolHandler;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.stream.ChunkedWriteHandler;
    import io.netty.util.concurrent.DefaultThreadFactory;
    import io.netty.util.internal.SystemPropertyUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PreDestroy;
    import javax.annotation.Resource;
    
    /** * @description: websocket 服务 * @author: wangzibin **/
    @Slf4j
    @Component
    public class WebsocketServer implements ApplicationListener<ContextRefreshedEvent> { 
    
        private static final int WEBSOCKET_IDLE_TIMEOUT = SystemPropertyUtil.getInt("star.websocket.idle_timeout", 180);
        private static final int WEBSOCKET_BACKLOG = SystemPropertyUtil.getInt("star.websocket.backlog", 200);
        private static final int WEBSOCKET_PORT = SystemPropertyUtil.getInt("star.websocket.port", 18680);
    
        private ServerBootstrap serverBootstrap;
        private EventLoopGroup bossGroup;
        private EventLoopGroup workerGroup;
        private Channel serverChannel;
    
        @Resource
        private CommonConnectHandler commonConnectHandler;
    
        @Resource
        private DiscardServerHandler discardServerHandler;
    
        @Override
        public void onApplicationEvent(ContextRefreshedEvent event) { 
            log.info("Starting WebSocketServer on port {} ", WEBSOCKET_PORT);
    
            long startMillis = System.currentTimeMillis();
            bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("WebsocketBoss"));
            workerGroup = new NioEventLoopGroup(0, new DefaultThreadFactory("WebsocketWorker"));
            serverBootstrap = new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    //使用NioServerSocketChannel作为服务器的通道实现
                    .channel(NioServerSocketChannel.class)
                    // 初始化服务器连接队列大小，服务端处理客户端连接请求是顺序处理的,所以同一时间只能处理一个客户端连接。
                    // 多个客户端同时来的时候,服务端将不能处理的客户端连接请求放在队列中等待处理
                    // 如果同一时刻超出队列上限则会拒绝连接
                    .option(ChannelOption.SO_BACKLOG, WEBSOCKET_BACKLOG)
                    // 禁用客户端 TCP Nagle 算法拥塞控制，这里需要实时性，不等待阻塞低载荷数据包的发送
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.TRACE))
                    .childHandler(new ChannelInitializer<SocketChannel>() { 
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception { 
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(64 * 1024));
                            pipeline.addLast(new ChunkedWriteHandler());
                            pipeline.addLast(new WebSocketServerProtocolHandler("/chatws"));
                            pipeline.addLast(new TextWebSocketFrameHandler());
                            pipeline.addLast(new MqttWebSocketCodec());
                            pipeline.addLast(commonConnectHandler);
                            pipeline.addLast(discardServerHandler);
                        }
                    });
    
            ChannelFuture channelFuture = serverBootstrap.bind(WEBSOCKET_PORT);
            channelFuture.syncUninterruptibly();
            serverChannel = channelFuture.channel();
            log.info("WebSocketServer started in {} ms", System.currentTimeMillis() - startMillis);
        }
    
    
        @PreDestroy
        public void close() { 
            log.info("Closing WebSocketServer");
            long startMillis = System.currentTimeMillis();
            try { 
                if (serverChannel != null) { 
                    serverChannel.close().syncUninterruptibly();
                }
            } catch (Throwable e) { 
                log.warn("serverChannel close error!", e);
            }
            try { 
                if (serverBootstrap != null) { 
                    bossGroup.shutdownGracefully();
                    workerGroup.shutdownGracefully();
                }
            } catch (Throwable e) { 
                log.warn("close eventLoopGroup error!", e);
            }
            log.info("WebSocketServer closed in {} ms", System.currentTimeMillis() - startMillis);
        }
    }

其他的没什么好说的，如果还陌生的同学请移步：[【如何构建商业级别聊天系统】开始搭建你的聊天服务 Netty 最佳实践快速入门][Netty]

不一样的在于我们在定制的 pipline 中添加的 handler 处理器，来重点看看他们都干了些什么

ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(64 * 1024));
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler("/chatws"));
    pipeline.addLast(new TextWebSocketFrameHandler());
    pipeline.addLast(new MqttWebSocketCodec());
    pipeline.addLast(commonConnectHandler);
    pipeline.addLast(discardServerHandler);

**HttpServerCodec**

首先是 HttpServerCodec 源码中写着这样一段注释

A combination of HttpRequestDecoder and HttpResponseEncoder which enables easier server side HTTP implementation. HttpRequestDecoder 和 HttpResponseEncoder 组合，使服务器端 HTTP 实现更容易

HttpServerCodec 为我们提供了开箱即用的 Http 服务的编解码具体的处理逻辑这里先不展开看了。

**HttpObjectAggregator**

对于 HttpObjectAggregator 官方给出如下解释：

A ChannelHandler that aggregates an HttpMessage and its following HttpContents into a single FullHttpRequest or FullHttpResponse (depending on if it used to handle requests or responses) with no following HttpContents. It is useful when you don't want to take care of HTTP messages whose transfer encoding is 'chunked'. Insert this handler after HttpResponseDecoder in the ChannelPipeline if being used to handle responses, or after HttpRequestDecoder and HttpResponseEncoder in the ChannelPipeline if being used to handle requests. ChannelPipeline p = ...;
        ...
        p.addLast("decoder", new HttpRequestDecoder());
        p.addLast("encoder", new HttpResponseEncoder());
        p.addLast("aggregator", new HttpObjectAggregator(1048576));
        ...
        p.addLast("handler", new HttpRequestHandler());
        
    For convenience, consider putting a HttpServerCodec before the HttpObjectAggregator as it functions as both a HttpRequestDecoder and a HttpResponseEncoder. Be aware that HttpObjectAggregator may end up sending a HttpResponse: 
    Response Status
    Condition When Sent
    100 Continue
    A '100-continue' expectation is received and the 'content-length' doesn't exceed maxContentLength
    417 Expectation Failed
    A '100-continue' expectation is received and the 'content-length' exceeds maxContentLength
    413 Request Entity Too Large
    Either the 'content-length' or the bytes received so far exceed maxContentLength
    请参阅:
    FullHttpRequest, FullHttpResponse, HttpResponseDecoder, HttpServerCodec

一种 ChannelHandler，它将 HttpMessage 及其后续 HttpContents 聚合为单个 FullHttpRequest 或 FullHttpResponse（取决于是否用于处理请求或响应），而不包含后续HttpContents。当您不想处理传输编码为“分块”的 HTTP 消息时，它非常有用。如果用于处理响应，则在ChannelPipeline 中的 HttpResponseCoder 之后插入此处理程序；如果用于处理请求，则在 ChannelPipeline 中的 HttpRequestDecoder 和 HttpResponseEncoder 之后插入此处理程序。
        ChannelPipeline p = ...;
        ...
        p.addLast("decoder", new HttpRequestDecoder());
        p.addLast("encoder", new HttpResponseEncoder());
        p.addLast("aggregator", new HttpObjectAggregator(1048576));
        ...
        p.addLast("handler", new HttpRequestHandler());
    为了方便起见，考虑将直接将 HttpServerCodec 放到 HttpObjectAggregator 之前，它既是 HttpRequestDecoder 又是 HttpResponseEncoder。
    请注意，HttpObjectAggregator 最终可能会发送 HttpResponse：
    响应状态
    发送时的条件
    100 Continue
    收到 “100 continue” 预期值，且 “content length” 不超过 maxContentLength
    417 Expectation Failed
    收到 “100 continue” 预期值，且 “content length” 超过 maxContentLength
    413 Request Entity Too Large
    “content-length” 或 迄今为止接收到的字节超过 maxContentLength
    请参阅:
    FullHttpRequest，FullHttpResponse，HttpResponseCoder，HttpServerCodec

而为什么我们要使用 HttpObjectAggregator 在 stack overflow 中作者 trustin 做出了如下回答 [how-to-use-httprequest-fullhttprequest-httpmessage-fullhttpmessage-an][]

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16][]

也就是说在 HTTP 消息经过 HttpObjectDecoder 的解码之后，该解码器会生成如下几个对象：

1.  一个 HttpRequest 或者 一个 HttpResponse
2.  一系列的 HttpContent，而最后的 HttpContent 是 LastHttpContent

也就是说我们通过 HttpServerCodec 进行解码后得到的这样一系列的对象导致我们需要对每个对象都进行特殊处理，一个典型的写法如下：

if (msg instanceof HttpRequest) { 
        ...
    }
    if (msg instanceof HttpContent) { 
        ...
        if (msg instanceof LastHttpContent) { 
            ...
        }
    }

而 FullHttpRequest 它实现了 HttpRequest、HttpContent 和 LastHttpContent。那么 HttpObjectAggregator 就帮助我们将以上三个聚合为了一个 FullHttpRequest 方便进行后续处理。（ps 作者也说了这样的设计不那么容易理解，但是后续好像也没改哈哈哈）

**ChunkedWriteHandler**

A ChannelHandler that adds support for writing a large data stream asynchronously neither spending a lot of memory nor getting OutOfMemoryError. Large data streaming such as file transfer requires complicated state management in a ChannelHandler implementation. ChunkedWriteHandler manages such complicated states so that you can send a large data stream without difficulties. To use ChunkedWriteHandler in your application, you have to insert a new ChunkedWriteHandler instance:
       ChannelPipeline p = ...;
       p.addLast("streamer", new ChunkedWriteHandler());
       p.addLast("handler", new MyHandler());
       
    Once inserted, you can write a ChunkedInput so that the ChunkedWriteHandler can pick it up and fetch the content of the stream chunk by chunk and write the fetched chunk downstream:
       Channel ch = ...;
       ch.write(new ChunkedFile(new File("video.mkv"));
       
    Sending a stream which generates a chunk intermittently
    Some ChunkedInput generates a chunk on a certain event or timing. Such ChunkedInput implementation often returns null on ChunkedInput.readChunk(ChannelHandlerContext), resulting in the indefinitely suspended transfer. To resume the transfer when a new chunk is available, you have to call resumeTransfer().

ChunkedWriteHandler 是一个添加支持大数据流文件异步写入的支持，既不消耗大量内存，也不获取 OutOfMemoryError。大数据流的传输例如文件传输在 ChannelHandler 的实现中需要需要管理复杂的状态。ChunkedWriteHandler 会对这些复杂状态进行管理，可以让我们毫不费力的进行大数据流传输。

**WebSocketServerProtocolHandler**

This handler does all the heavy lifting for you to run a websocket server. It takes care of websocket handshaking as well as processing of control frames (Close, Ping, Pong). Text and Binary data frames are passed to the next handler in the pipeline (implemented by you) for processing. See io.netty.example.http.websocketx.html5.WebSocketServer for usage. The implementation of this handler assumes that you just want to run a websocket server and not process other types HTTP requests (like GET and POST). If you wish to support both HTTP requests and websockets in the one server, refer to the io.netty.example.http.websocketx.server.WebSocketServer example. To know once a handshake was done you can intercept the ChannelInboundHandler.userEventTriggered(ChannelHandlerContext, Object) and check if the event was instance of WebSocketServerProtocolHandler.HandshakeComplete, the event will contain extra information about the handshake such as the request and selected subprotocol.

WebSocketServerProtocolHandler 这个处理程序将完成 websocket 服务器所有需要进行的繁重工作。它负责 websocket 握手以及控制帧（Close、Ping、Pong）的处理。文本和二进制数据帧被传递到管道中的下一个处理程序（由您实现）进行处理。

也就是说此 Handler 会帮我们处理 websocket 握手，关闭，ping，pong 等操作，而真正的业务数据帧会传递给我们后续的 Handler 进行处理，你只需要专注于业务逻辑的处理即可。

感兴趣的同学可以参考一下 netty 给出的例子  
[io.netty.example.http.websocketx.server.WebSocketServer][]

/* * Copyright 2012 The Netty Project * * The Netty Project licenses this file to you under the Apache License, * version 2.0 (the "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at: * * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * License for the specific language governing permissions and limitations * under the License. */
    package io.netty.example.http.websocketx.server;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    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.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
    import io.netty.handler.ssl.SslContext;
    
    /** */
    public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> { 
    
        private static final String WEBSOCKET_PATH = "/websocket";
    
        private final SslContext sslCtx;
    
        public WebSocketServerInitializer(SslContext sslCtx) { 
            this.sslCtx = sslCtx;
        }
    
        @Override
        public void initChannel(SocketChannel ch) throws Exception { 
            ChannelPipeline pipeline = ch.pipeline();
            if (sslCtx != null) { 
                pipeline.addLast(sslCtx.newHandler(ch.alloc()));
            }
            pipeline.addLast(new HttpServerCodec());
            pipeline.addLast(new HttpObjectAggregator(65536));
            pipeline.addLast(new WebSocketServerCompressionHandler());
            pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
            pipeline.addLast(new WebSocketIndexPageHandler(WEBSOCKET_PATH));
            pipeline.addLast(new WebSocketFrameHandler());
        }
    }

#### 三、定制自己的数据处理 Handler ####

**TextWebSocketFrameHandler**

import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
    import io.netty.handler.ssl.SslHandshakeCompletionEvent;
    import lombok.extern.slf4j.Slf4j;
    
    @Slf4j
    public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { 
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) { 
            log.info("text web socket text info {}",textWebSocketFrame.text());
            ctx.fireChannelRead(textWebSocketFrame.text());
        }
    }

这里我们仅接受 TextWebSocketFrame 类型的数据，由上个 WebSocketServerProtocolHandler 处理完成，这里直接获取其文本数据。

**MqttWebSocketCodec**

import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToMessageCodec;
    import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.List;
    
    @Slf4j
    public class MqttWebSocketCodec extends MessageToMessageCodec<BinaryWebSocketFrame, ByteBuf> { 
        @Override
        protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) { 
            log.info("encode ctx={}, msg={}, out={}", ctx, msg, out);
            out.add(new BinaryWebSocketFrame(msg.retain()));
        }
    
        @Override
        protected void decode(ChannelHandlerContext ctx, BinaryWebSocketFrame msg, List<Object> out) { 
            log.info("decode ctx={}, msg={}, out={}", ctx, msg, out);
            out.add(msg.retain().content());
        }
    }

MqttWebSocketCodec 为什么起这个名字呢？这里留个坑，为后续建立与 WebSocket 协议上的 Mqtt 协议打个基础。这里只走到了其 encode 方法，将数据写回给客户端，TextWebSocketFrame 处理传入的从类型来看 decode 是没走到的。

还有两个 Handler ：CommonConnectHandler 和 DiscardServerHandler 是连接处理和数据分发的两个定制 Handler ，限于篇幅放到下篇来讲。下面做个简单的总结

#### 总结 ####

对于构建 WebSocket 服务，Netty 为我们提供了开箱即用的 Handler ，我们只需要按一定顺序组织这些 Handler，就可以实现一个复杂的 WebSocket 服务器了，利用这些开箱即用的组件我们可以吧更多的精力集中于业务逻辑上。

常规的，构建 WebSocket 服务需要以下 Handler

1.  **HttpServerCodec** ： Http 编解码器，用于建立 Http 服务，对 Http Message 进行编解码
2.  **HttpObjectAggregator**: HttpObject 聚合处理器，Netty 对于 Http 消息区分了 HttpRequest、HttpResponse 、HttpContent 、LastHttpContent 等，该处理器帮助我们将 HttpServerCodec 编解码出来的 一系列 HttpObject 类聚合成统一的 FullHttpMessage 方便后续处理。
3.  **ChunkedWriteHandler**: 开箱即用的大数据流传输处理工具，让我们的服务直接支持复杂的大数据流处理功能。
4.  **WebSocketServerProtocolHandler**: WebSocket 协议升级握手处理器，处理 WebSocket 握手流程、关闭、Ping、Pong 等操作，将其他文本等业务数据传到后续自定义的 Handler 进行处理。

本篇为 WebScoket 服务构建上篇，算是对整个 WebSocket 服务开了一个头，对整体的协议以及协议处理流程做了一个梳理和介绍。下篇将着重讲解，连接以及数据的处理。当然对于商业级别的服务这仅仅是个开始，让我们一起进行迭代吧。

我是 **dying 搁浅**，关注我，不迷路，我们下篇文章再见。

[rfc6455]: https://datatracker.ietf.org/doc/html/rfc6455
[WebSocket]: https://blog.csdn.net/w903328615/article/details/118442728
[Netty]: https://blog.csdn.net/w903328615/article/details/120239503
[how-to-use-httprequest-fullhttprequest-httpmessage-fullhttpmessage-an]: https://stackoverflow.com/questions/29213793/how-to-use-httprequest-fullhttprequest-httpmessage-fullhttpmessage-an
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/f159b42724404506a55c7a8bcbf9e4d8.png
[io.netty.example.http.websocketx.server.WebSocketServer]: https://github.com/netty/netty/blob/4.1/example/src/main/java/io/netty/example/http/websocketx/server/WebSocketServer.java