月更干货：基于netty的websocket策略消息路由

╰+哭是因爲堅強的太久メ 2023-01-01 04:49 161阅读 0赞

> 该文主要写个人的例子设计实现思路，如觉得部分知识讲得不够细得建议自行搜索，也可参考个人之前的[Web体系结构之Reactor线程模型知识整理][Web_Reactor]文章。

# 前言 #

又到了月更时间了，本来想再把Websocket拖一拖看看能不能写其它内容的，但烂大街的内容不想写，觉得烂大街的内容学习知识梳理写到自己的静态博客上就行了，拖到现在实在想不到有什么好的其它内容可以写，只好接上月底的策略模式续写WebSocket聊天室例子了。该文章干货为主，湿货为辅，本年度本搬砖工最后一篇文章，用于这万恶的一年里最后一个工作日的工作收尾(摸鱼党的骄傲)。

# WebSocket简介 #

WebSocket是一种在单个TCP连接上进行全双工通信的协议，在WebSocket API中，浏览器和服务器只需要通过Http协议101状态码进行一次握手，两者之间就直接可创建持久性的连接，并进行双向数据传输(百科简介)。形象点的比喻就是语音通信与语音信息的区别，语音通信只要其中一方不关闭双方就可以一直随便BB下去，语音信息一般则是你一句我一句。  
同是协议，WebSocket与Http的主要区别在于一个是持久性的连接，一个是一次性连接，从而显出了WebSocket的以下优点：

*  实时性：客户端与服务端可随时实时发送消息，常见的应用实践常见如聊天室、服务器消息推送
 *  减少开销：相比使用Http进行长轮询消息推送减少了更多的开销
 *  …

为了减少篇幅，以下便列举该文例子的WebSocket连接建立图作罢：  
![WebSocket连接图][WebSocket]  
可以看出进行建立WebSocket连接时客户端会发一个101的Http状态码，Http 101状态码指请求切换协议，且只能切换到更高级的协议，该例中就是切换到websocket协议(Response Header中的upgrade)。  
**Websocket客户端部分API：**

// 创建WebSocket实例
    let socket = new WebSocket("ws://localhost:9000/chat");
    // 获取websocket状态，含CONNECTING、OPEN、CLOSING、CLOSED四种，
    // 可用WebSocket.Xxx常量进行websocket状态判断
    let state = socket.readyState;
    socket.onopen = function(event) {
      // 连接成功后回调
    }
    socket.onmessage = function (event) {
      // 收到服务端消息后回调
    }
    socket.onclose = function (event) {
        console.log("websocket close"); //关闭后回调
    }
    // 向服务端发送消息
    socket.send('hello world');

由于是前端渣渣就不再献丑了。

# Netty #

都说Netty是一款高性能的网络应用程序框架，高性能应用常见的选项，后端WebSocket实现的最常见、常用方式，但问题是Netty是怎么实现的？在此之前个人认为需要先了解下**Web请求处理体系结构、I/O多路复用、Reactor(响应式)线程模型** 这3个知识点。

## Netty前置知识 ##

> *该部分主要源自学习时搜到的知识整理，如有侵犯请告知。*

### Web请求处理体系结构 ###

每个Web应用的使用看起来都是你请求Web回应，但其实请求的处理结构中其实又可以分为以下两种：

*  **thread-based architecture，基于线程的处理结构**  
    基于线程的体系结构通常会使用多线程来处理客户端的请求，每当接收到一个请求，便开启一个独立的线程来处理，该结构也可称作传统的Web请求处理结构。
 *  **event-driven architecture，事件驱动的处理结构**  
    事件驱动的体系结构是目前比较广泛使用的一种，该方式会定义一系列的事件处理器来响应事件的发生，并且将服务端接受连接与对事件的处理分离。其中，事件是一种状态的改变，如TCP中socket的new incoming connection、ready for read、ready for write。

### I/O多路复用 ###

**I/O多路复用**可以用网络编程从字面意思简单的去解释：`I/O`一般指网络`I/O`，多路指多个TCP连接，复用指重复使用一个或少量线程，连起来就是多个(描述符-fd)`I/O`通过重复使用一个线程去处理多个连接。一般`I/O多路复用`机制都依赖于一个事件多路分离器(Event Demultiplexer)，分离器对象可将来自事件源的`I/O`事件分离出来，并分发到对应的read/write事件处理器(Event Handler)。这个事件多路分离器看起来有点眼熟，个人认为其实Web事件驱动的处理结构与Netty都只是`I/O多路复用`的一种实现体现。  
看起来很牛逼，但牛逼的功能都是需要底层操作系统去支持的，目前Linux支持`I/O多路复用`的系统调用模块有`select`、`poll`、`epoll`，篇幅原因…

### Reactor(响应式)线程模型 ###

Reactor线程模型是一种事件处理模式，用于处理由一个或多个输入并发传递给服务处理程序的服务请求，服务处理程序将传入的请求分解，并将它们同步地分派给相关的请求处理程序。Reactor线程模型是Web事件驱动的处理结构(event-driven architecture)的一种实现模型，被广泛用于基于`I/O多路复用机制`设计实现的软件编程中，如Netty、Redis都使用该模式解决高性能并发问题。Reactor模型主要分为以下三个角色：

*  Reactor：将I/O事件分配给对应的handler处理
 *  Acceptor：处理客户端连接事件
 *  Handler：将自身与事件绑定，处理非阻塞任务  
    常用的Reactor线程模型有3种：单Reactor单线程模型、单Reactor多线程模型、主从多线程模型，Netty作为该线程模型的实现框架自然也是提供以上3种模型的设置的。下图为主从多线程模型的Reactor模型图：  
    ![multi-reactor-multi-thread.jpeg][]

## Netty简介 ##

Netty是一个异步事件驱动的网络应用程序框架，用于快速开发可维护的高性能协议服务器和客户端（*摘自[官网][Link 1]* ），且结合了许多协议，大大精简了开发人员耗费在网络编程上的时间。至于Netty的实现原理相信看了前面的前置知识都有一个大概的轮廓了，就是一个基于被广泛用于基于`I/O多路复用机制`设计实现的软件编程中的**Reactor线程模型**的NIO(非阻塞)网络通信框架，常用于基于Web事件驱动体系结构的应用实现，比如阿里的RocketMQ为什么这么高性能？就是因为使用了Netty(rocketmq-remoting中的netty client类):  
![NettyRemotingClient][]

### Netty核心组件 ###

原生Java NIO是NIO了，但极其不易用，需要开发者了解大量的网络编程知识，代码编写复杂，而Netty通过对NIO的封装极大地简化和简化了网络编程，使开发者可以快速上手网络编程。在将例子前围绕`ChannelPipeline`与**线程模型**简单介绍下Netty的核心组件(主要参考自源码文档)。

#### `ChannelPipeline`事件处理涉及到的组件 ####

<table> 
 <thead> 
  <tr> 
   <th>组件</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Channel</td> 
   <td>网络套接字的连接，或能进行读、写、连接和绑定等I/O操作的组件连接，该连接可设置相应的参数配置(如缓冲区)与I/O操作(如读写)</td> 
  </tr> 
  <tr> 
   <td>ChannelHandler</td> 
   <td>处理<code>I/O</code>事件或拦截<code>I/O</code>操作，并将事件转发到其在<code>ChannelPipeline</code>中的下一个handler。其中<code>I/O</code>事件又可划分为入站事件与出站事件(如常见的进站解码出站编码)，对应的处理器分别为<code>ChannelInboundHandler</code>与<code>ChannelOutboundHandler</code>,官方建议实现适配器类<code>ChannelInboundHandlerAdapter</code>与<code>ChannelOutboundHandlerAdapter</code>进行事件处理</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext</td> 
   <td>负责各个<code>ChannelHandler</code>在<code>ChannelPipeline</code>中与其它handler的交互，转发事件到下一个handler。</td> 
  </tr> 
  <tr> 
   <td>ChannelPipeline</td> 
   <td>可看成是netty事件处理的容器，一个包含了事件所需的<code>ChannelHandler</code>列表，用于处理或拦截<code>Channel</code>入站和出站事件操作。每个<code>Channel</code>都有属于自己的<code>ChannelPipeline</code>，当一个<code>Channel</code>被创建时<code>ChannelPipeline</code>也会随着创建。</td> 
  </tr> 
 </tbody> 
</table>

为了更形象的展示以上组件之间的关系，截了以下`ChannelPipeline`的注释文档图：  
![ChannelPipeline][]

#### Reactor线程模型在Netty中的组件体现 ####

<table> 
 <thead> 
  <tr> 
   <th>组件</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Bootstrap</td> 
   <td>引导<code>Channel</code>以供客户端使用</td> 
  </tr> 
  <tr> 
   <td>ServerBootstrap</td> 
   <td>引导<code>Channel</code>子类<code>ServerChannel</code>以供服务端使用</td> 
  </tr> 
  <tr> 
   <td>EventExecutorGroup接口</td> 
   <td>继承了<code>ScheduledExecutorService</code>及<code>Iterator</code>,一个基于netty事件循环处理机制而实现的线程池抽象,遍历返回为其子类<code>EventExecutor</code>(一个事件处理器)</td> 
  </tr> 
  <tr> 
   <td>EventLoopGroup接口</td> 
   <td>继承了<code>EventExecutorGroup</code>，提供了<code>Channel</code>的注册，且重写遍历方法<code>next()</code>的返回值为<code>EventLoop</code></td> 
  </tr> 
  <tr> 
   <td>EventLoop接口</td> 
   <td>继承了<code>EventLoopGroup</code>接口，用于处理Channel连接生命周期中所发生的事件。当<code>Channel</code>在一个<code>EventLoop</code>上注册了，<code>EventLoop</code>会负责处理该<code>Channel</code>上的所有<code>I/O</code>操作，一般一个<code>EventLoop</code>实例负责处理多个<code>Channel</code>。</td> 
  </tr> 
 </tbody> 
</table>

以下为Channel、EventLoop、Thread及EventLoopGroup之间的关系(摘自《Netty实战》)：  
![channel-relation][]

*  一个EventLoopGroup包含一个或多个EventLoop
 *  一个EventLoop在它的生命周期内只和一个Thread绑定
 *  所有由EventLoop处理的I/O事件都将在它专有的Thread上被处理
 *  一个Channel在它的生命周期内只注册一个EventLoop
 *  一个EventLoop可能被分配给一个或多个Channel

在Reactor线程模型中主要有Reactor(I/O事件分发)、Acceptor(客户端连接事件处理)、Handler(绑定事件处理非阻塞的任务)三个角色，相信通过以上的介绍中我们可以大致找到与Reactor线程模型中角色对应的组件了：

*  `EventLoopGroup`：基于netty中`Channel`操作的线程池抽象，既是Reactor，也是Acceptor(Reactor与Acceptor实质即不同的线程负责特定的工作)
 *  `ChannelHandler`：相当于Reactor中的Handler，为了方便Handler之间的交互与事件处理，netty添加了`ChannelPipeline`、`ChannelHandlerContext`两个角色
 *  `Bootstrap`与`ServerBootstrap`就简单的看成是client与server就好了

# 基于Netty搭建的WebSocket例子 #

终于码到实例部分了，终于可以CV操作+点设计思路描述完事了。该例子只是一个简单的基于Netty搭建的聊天室，纯粹是工作需要所以拿出了1年前写的渣渣demo学习并血洗，学习完并不影响我前端依旧烂的显示。该例子涉及到的知识点：

*  Spring Boot
 *  基于Spring实现的策略模式(伪·无策略模式)进行消息分发
 *  Netty

## 实现思路 ##

### netty实现思路 ###

Netty结合了许多协议，对于所有的事件进出站处理都是交给Handler的，所以当使用netty作为服务端WebSocket的实现时我们只需了解netty对WebSocket的一些封装与进出站处理即可，开箱即用，无需像Java NIO那样什么都要去考虑实现。  
Netty提供了`ChannelInboundHandlerAdapter`接口的简单抽象实现类`SimpleChannelInboundHandler<I>`，该处理器只处理仅特定类型的消息，泛型`<I>`为消息的类型，开发者如果想简单实现处理特定消息类型的处理器只需继承该类并添加到`ChannelPipeline`即可。  
Netty提供了`WebSocketFrame`抽象类作为所有WebSocket数据帧封装的基类，其提供了以下实现类：  
![WebSocketFrame-diagram.png][]  
为了简单处理`SimpleChannelInboundHandler`消息类型选取了文本帧`TextWebSocketFrame`，该类的`text()`方法会以`UTF-8`的字符串格式获取WebSocket的传输内容，每次server与client的WebSocket交互都以字符串传输即可。

### 消息分发实现思路 - 策略模式 ###

虽然消息都是以字符串传送，但总得有个分发机制，不然明明私聊的消息也被发到群聊可能就炸了，于是我就定了以JSON字符串的格式进行传输，且JSON中需带有必需的参数用于消息类型判断与分发。  
消息类型判断处理如果每次都用`if`写起来太难看了，业务多起来时也不好处理，更不符合我为自己编码的逼格，然后想出了一套伪无策略模式的实现思路：

1.  WebSocket消息类遵循特定后缀约束，且都继承同一父类
2.  项目初始化时扫描所有WebSocket消息的策略处理类
3.  通过反射获取策略处理类消息处理方法的参数类型，截取后缀前的字符串作为策略名
4.  将策略名与策略处理类映射成Map，策略名与对应消息类映射成Map，暴露一个策略名列表API以供查询
5.  WebSocket握手时将参数添加到地址栏，服务端接收到握手消息时将地址栏参数转换成对应消息对象交给策略Context，Context从Map中获取对应的策略类处理消息然后完成握手
6.  客户端传输WebSocket JSON消息传输到服务端后，Context将该JSON转换为相应的消息类并交给对应的策略类处理  
    于是有了以下握手消息处理序列图：  
    ![handshake sequence][]  
    client发送消息处理序列图：  
    ![channelRead0][]

## 代码样例 ##

### WebSocket消息入站处理器 - `ChatMsgInboundHandler` ###

@Component
    @ChannelHandler.Sharable
    @Slf4j
    @AllArgsConstructor
    public class ChatMsgInboundHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
        private final WebSocketConfig webSocketConfig;
        private final MessageContext messageContext;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // WebSocket通过Http握手建立起长连接
            if (msg instanceof FullHttpRequest) {
                FullHttpRequest request = (FullHttpRequest) msg;
                // 提取地址栏参数
                JSONObject paramsJson = RequestUtils.urlParamsToJson(request.uri());
                // 清空参数重置路径，故不能与上一行提取互换
                httpRequestHandle(ctx, request);
                // 将地址栏参数转换为json
                WebSocketMessage message = messageContext.convertJsonToMessage(paramsJson);
                message.setChannel(ctx.channel());
                log.info("user {} is online", message.getFromUser());
                messageContext.registerMessage(message);
            }
            super.channelRead(ctx, msg);
        }
    
        /**
         * 处理连接请求，客户端WebSocket发送握手包时会执行这一次请求
         *
         * @param ctx
         * @param request
         */
        private void httpRequestHandle(ChannelHandlerContext ctx, FullHttpRequest request) {
            String uri = request.uri();
            // 判断配置的websocket contextPath与请求地址中的contextPath是否一致
            if (webSocketConfig.getContextPath().equals(RequestUtils.getBasePath(uri))) {
                // 因为有可能携带了参数，导致客户端一直无法返回握手包，因此在校验通过后，重置请求路径
                request.setUri(webSocketConfig.getContextPath());
            } else {
                ctx.close();
            }
        }
    
        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            messageContext.removeChannel(ctx.channel());
            log.info("channelUnregistered: {}", ctx.channel().id().asLongText());
            super.channelUnregistered(ctx);
        }
    
        /**
         * 消息处理
         *
         * @param ctx
         * @param frame
         */
        @Override
        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
            messageContext.handleMessage(frame.text());
        }
    
        /**
         * 对消息处理过程中抛出的异常进行处理
         *
         * @param ctx
         * @param cause
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            if (cause instanceof BusinessException) {
                System.out.println(ctx.channel().isOpen());
                ServerResponse<?> response = ServerResponse.serverError(cause.getMessage());
                log.error("netty handler exceptionCaught: {}", cause.getMessage());
                ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(response)));
            } else {
                super.exceptionCaught(ctx, cause);
            }
        }
    }

方法简介：

*  `channelRead()`：`Channel`接收到消息时的处理方法
 *  `channelRead0()`：定义于`SimpleChannelInboundHandler<I>`类中，每次接收到`<I>`类型消息时都会调用该方法
 *  `channelUnregistered()`:`ChannelHandlerContext`中的`Channel`已从其`EventLoop`中取消注册
 *  `exceptionCaught()`：处理handler抛出的异常  
    由于使用了策略模式所以不同消息处理的业务复杂性都分别封装到相应的策略处理类中了，而策略类的调度则封装到`MessageContext`中，这让Handler看起来十分清爽。`ChatMsgInboundHandler`的`channelRead()`方法结尾调用父类方法是为了让消息进入到`channelRead0()`中处理，一般也是提倡继承了`SimpleChannelInboundHandler<I>`的类对`<I>`类型消息的处理置于`channelRead0()`中，主要因`SimpleChannelInboundHandler<I>`的部分方法源码如下：

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        boolean release = true;
        try {
            if (acceptInboundMessage(msg)) {
                @SuppressWarnings("unchecked")
                I imsg = (I) msg;
                channelRead0(ctx, imsg);
            } else {
                release = false;
                ctx.fireChannelRead(msg);
            }
        } finally {
            if (autoRelease && release) {
                ReferenceCountUtil.release(msg);
            }
        }
    }
    
    protected abstract void channelRead0(ChannelHandlerContext ctx, I msg) throws Exception;

### 策略核心上下文 - `MessageContext` ###

@Slf4j
    @SuppressWarnings({"rawtypes", "unchecked"})
    @Component
    public class MessageContext implements ApplicationContextAware {
        /**
         * Map{消息类名前缀：消息类Class}，用于fastjson根据消息内容中的msgType反序列化成实际WebSocket消息类
         */
        private Map<String, Class<? extends WebSocketMessage>> msgTypeMap;
        /**
         * Map{消息类名前缀：消息类对应的handler}，消息类名前缀=策略名，用于根据消息类型参数获取对应的消息策略处理器
         */
        private Map<String, MessageHandler> messageHandlerMap;
    
        /**
         * 初始化：注入灵魂，映射初始化
         */
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            // 获取容器中的所有MessageHandler
            Map<String, MessageHandler> handlerBeanMap = applicationContext.getBeansOfType(MessageHandler.class);
            this.messageHandlerMap = new HashMap<>(handlerBeanMap.size());
            this.msgTypeMap = new HashMap<>(handlerBeanMap.size());
            // 反射获取所有策略处理类MessageHandler实际处理的消息类型，并生成所需Map:msgTypeMap、messageHandlerMap
            handlerBeanMap.values()
                    .forEach(messageHandler -> {
                        Method handleMsg;
                        try {
                            handleMsg = ReflectUtils.method(messageHandler, true, "handleMsg",
                                    method -> method.getParameterCount() == 1 && !Objects.equals(method.getParameterTypes()[0], WebSocketMessage.class));
                        } catch (ClassNotFoundException | NoSuchMethodException e) {
                            e.printStackTrace();
                            log.error("MessageContext message handler map init failed(handler={}",messageHandler.getClass().getSimpleName());
                            return;
                        }
                        Class<? extends WebSocketMessage> msgClass = (Class<? extends WebSocketMessage>) handleMsg.getParameterTypes()[0];
                        String msgClassName = msgClass.getSimpleName();
    
                        String msgType = getMsgType(msgClassName);
                        this.messageHandlerMap.put(msgType, messageHandler);
                        this.msgTypeMap.put(msgType, msgClass);
                    });
            log.info("websocket message context init completed, msg type: {}, handler map: {}", msgTypeMap, messageHandlerMap);
        }
    
        /**
         * 根据消息类Class.simpleName获取实际策略名，如PrivateChatWebSocketMessage策略名为PrivateChat
         *
         * @param msgClassName
         * @return
         */
        private String getMsgType(String msgClassName) {
            return msgClassName.contains(WebSocketMessage.MSG_TYPE_SEPARATOR) ?
                    StringUtils.substringBefore(msgClassName, WebSocketMessage.MSG_TYPE_SEPARATOR) : msgClassName;
        }
        
        public ServerResponse handleMessage(String msgJson) {
            if (JSONValidator.from(msgJson).validate()) {
                // 将消息转换为对应 WebSocketMessage 子类
                JSONObject jsonObject = JSON.parseObject(msgJson);
                WebSocketMessage message = convertJsonToMessage(jsonObject);
                // 根据获取消息类型获取消息处理器处理消息
                String msgType = jsonObject.getString(WebSocketMessage.MSG_TYPE);
                MessageHandler messageHandler = getMessageHandler(msgType);
                return messageHandler.handleMsg(message);
            } else {
                throw new IllegalArgumentException("invalid msg json");
            }
        }
        
        /**
         * @param msgJson
         * @param <T>
         * @return
         */
        public <T extends WebSocketMessage> WebSocketMessage convertJsonToMessage(JSONObject msgJson) {
            String msgType = msgJson.getString(WebSocketMessage.MSG_TYPE);
            Assert.isTrue(msgTypeMap.containsKey(msgType), "Unknown json msgType " + msgType);
            return msgJson.toJavaObject(msgTypeMap.get(msgType));
        }
        
        // 省略部分......
        
    }

由于策略名与策略处理类都是在该Context中动态初始化的，所以当有新的消息策略时只需添加新的消息类(特定后缀命名)与策略处理类，策略名与处理类匹配交给Context即可。

### WebSocket消息抽象类 - `WebSocketMessage` ###

@Data
    @Accessors(chain = true)
    public abstract class WebSocketMessage implements Serializable {
        public static final String MSG_TYPE = "msgType";
        public static final String MSG_TYPE_SEPARATOR = "WebSocket";
        public static final String MSG_PATTERN = "%s: %s";
    
        private String msgType;
        protected String fromUser;
        protected String content;
        @JsonIgnore
        protected Channel channel;
    
        public String userMsg() {
            return String.format(MSG_PATTERN, fromUser, content);
        }
    
    }

### 策略处理接口 - `MessageHandler` ###

public interface MessageHandler<T extends WebSocketMessage> {
        /**
         * handle message by corresponding handler
         * @param msg
         * @return
         */
        ServerResponse<?> handleMsg(T msg);
    
        /**
         * register channel from handler
         * @param msg
         */
        void registerChannel(T msg);
    
        /**
         * remove channel from handler
         * @param channel
         */
        void removeChannel(Channel channel);
    
    }

### 群聊消息策略处理类 - `GroupChatMessageHandler` ###

@Component
    public class GroupChatMessageHandler implements MessageHandler<GroupChatWebSocketMessage>{
        /**
         * {roomId:ChannelGroup}映射，ChannelGroup维护进入了该聊天室的所有用户channel
         */
        private final Map<String, ChannelGroup> roomChannelMap = new ConcurrentHashMap<>();
    
        @Override
        public ServerResponse<?> handleMsg(GroupChatWebSocketMessage msg) {
            ChannelGroup roomGroup = roomChannelMap.get(msg.getRoomId());
            // ... 消息DB存储
            ServerResponse<String> response = ServerResponse.success(msg.userMsg());
            roomGroup.writeAndFlush(WebSocketMessageUtils.websocketFrame(msg));
            return response;
        }
    
        @Override
        public void registerChannel(GroupChatWebSocketMessage msg) {
            roomChannelMap.compute(msg.getRoomId(), (key, group) -> {
                if (group == null) {
                    group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
                }
                group.add(msg.getChannel());
                return group;
            });
        }
    
        @Override
        public void removeChannel(Channel channel) {
            roomChannelMap.values()
                    .forEach(channels -> channels.remove(channel));
        }
    }

**注：channel.writeAndFlush(Object)的参数需为TextWebSocketFrame实例**

### Spring Netty配置,服务端引导类配置 ###

#### 配置类`WebSocketConfig` ####

@Slf4j
    @Configuration
    public class WebSocketConfig {
        @Getter
        @Value("${netty.server.port:9000}")
        private Integer port;
        @Getter
        @Value("${netty.websocket.path:/chat}")
        private String contextPath;
    
        private NioEventLoopGroup bossGroup;
        private NioEventLoopGroup workerGroup;
    
        @Bean
        public NioEventLoopGroup bossGroup() {
            return bossGroup = new NioEventLoopGroup();
        }
    
        @Bean
        public NioEventLoopGroup workerGroup() {
            return workerGroup = new NioEventLoopGroup();
        }
    
        @Bean
        public ServerBootstrap serverBootstrap(NioEventLoopGroup bossGroup, NioEventLoopGroup workerGroup, ChatServerInitializer chatServerInitializer) {
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    // boss负责接收客户端的tcp连接请求,worker负责与客户端的事件于I/O处理
                    .group(bossGroup, workerGroup)
                    //配置客户端的channel类型
                    .channel(NioServerSocketChannel.class)
                    .childHandler(chatServerInitializer)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
            serverBootstrap.bind(port);
            log.info("netty start on port: {}", port);
            // 绑定I/O事件的处理类,WebSocketChildChannelHandler中定义
            return serverBootstrap;
        }
    
        @PreDestroy
        public void destroy() {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            log.info("netty shutdown gracefully");
        }
    }

#### 运行类`ChatApplication` ####

@SpringBootApplication(scanBasePackageClasses = WebSocketConfig.class, exclude = DataSourceAutoConfiguration.class)
    //@MapperScan(basePackageClasses = UserInfoMapper.class)
    public class ChatApplication {
        public static void main(String[] args) {
            SpringApplication.run(ChatApplication.class, args);
        }
    }

该例子中去除了数据库的配置，如进行数据库测试把`@SpringBootApplication`的参数配置与`@MapperScan`前的注释符去掉配置本地数据库添加自己需要的操作即可。  
注：如果要在Handler中需要注入单例，则需对Handler添加netty中的`@ChannelHandler.Sharable`注解，用于标识一个Handler可以被添加到多个`ChannelPipeline`中，否则netty会因把单例注入到Handler中而报`ChannelPipelineException`。

## 效果演示 ##

### 私聊演示 ###

![private-chat][]

### 群聊演示 ###

![group-chat][]

# 结语 #

该文章主要内容为个人基于netty进行聊天室例子的WebSocket实现，同时码了一些自己对netty知识的简单梳理与总结，所以也不会对netty做很全面的介绍(如`ByteBuf`、零拷贝等)。如果要做服务端实时推送的可以参考该文例子的群聊推送实现，将不同的用户`Channel`分组即可实现指定用户组消息推送。由于只是一个简单的例子，所以请忽略设计的一些细节问题，如握手异常时的处理、`Channel`的维护(例中遍历移除当用户多时会存在性能问题)等。

该文中梳理的知识主要来自以下3个部分：

*  网络知识整理，总结到[Web体系结构之Reactor线程模型知识整理][Web_Reactor]文章中以供参考
 *  《Netty实战》
 *  netty源码

这样写维持月更都好难啊啊啊啊啊啊啊啊啊啊啊啊啊啊，明年一月写啥好。。。

# 附 #

代码地址：[github：Wilson-He/netty-simple-chat][github_Wilson-He_netty-simple-chat]  
如果对个人策略模式的演变与实现思路感兴趣的可以看我上月文章：[Spring+策略模式=无策略?  
][Spring_](1个月前写的代码跟一个月后写的代码区别是很大的，就算别人问我一周前写的代码我都会一脸懵逼，所以不要问我以前代码的细节问题)  
![unknow][]

[Web_Reactor]: https://blog.csdn.net/z28126308/article/details/111870718
[WebSocket]: /images/20221120/4ebd1dfa5c6e49c593b6c34a4bf598ef.png
[multi-reactor-multi-thread.jpeg]: https://img-blog.csdnimg.cn/img_convert/96a2075f295f273bff3c1405ddf5bc85.png
[Link 1]: https://netty.io/
[NettyRemotingClient]: https://img-blog.csdnimg.cn/img_convert/8c3c579b055414dbc4cefbf9a8b341d9.png
[ChannelPipeline]: https://img-blog.csdnimg.cn/img_convert/d647b25687ffe8bb3aacc3611bb78c35.png
[channel-relation]: https://img-blog.csdnimg.cn/img_convert/42a1a234e9cd315546fa150cfa074e80.png
[WebSocketFrame-diagram.png]: https://img-blog.csdnimg.cn/img_convert/94ff90325001e5f87a532ba90180dab7.png
[handshake sequence]: https://img-blog.csdnimg.cn/img_convert/572e2741b79ce833dab5e7e37b36a0e0.png
[channelRead0]: https://img-blog.csdnimg.cn/img_convert/99c6799203726ce50536cd5af8ec2ff5.png
[private-chat]: https://img-blog.csdnimg.cn/img_convert/9b8781a589a255ad384c1fe1ff4289f1.gif
[group-chat]: https://img-blog.csdnimg.cn/img_convert/3279264d28260a1f34904cf1ed8b3b76.gif
[github_Wilson-He_netty-simple-chat]: https://github.com/Wilson-He/netty-simple-chat
[Spring_]: https://juejin.cn/post/6900725430236479496
[unknow]: https://img-blog.csdnimg.cn/img_convert/f7e84fe0e371bfac6054be61d121ac5d.png