前言

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

WebSocket简介

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

同是协议，WebSocket与Http的主要区别在于一个是持久性的连接，一个是一次性连接，从而显出了WebSocket的以下优点：

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

为了减少篇幅，以下便列举该文例子的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模型图：Netty简介

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

Netty核心组件

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

ChannelPipeline事件处理涉及到的组件

Channel：网络套接字的连接，或能进行读、写、连接和绑定等I/O操作的组件连接，该连接可设置相应的参数配置(如缓冲区)与I/O操作(如读写)

ChannelHandler：处理I/O事件或拦截I/O操作，并将事件转发到其在ChannelPipeline中的下一个handler。其中I/O事件又可划分为入站事件与出站事件(如常见的进站解码出站编码)，对应的处理器分别为ChannelInboundHandler与ChannelOutboundHandler,官方建议实现适配器类ChannelInboundHandlerAdapter与ChannelOutboundHandlerAdapter使用

ChannelHandlerContext负责各个ChannelHandler在ChannelPipeline中与其它handler的交互，转发事件到下一个handler。

ChannelPipeline：可看成是netty事件处理的容器，一个包含了事件所需的ChannelHandler列表，用于处理或拦截Channel入站和出站事件操作。每个Channel都有属于自己的ChannelPipeline，当一个Channel被创建时ChannelPipeline也会随着创建。

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

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

Bootstrap：引导Channel以供客户端使用

ServerBootstrap：引导Channel子类ServerChannel以供服务端使用

EventExecutorGroup接口 ：继承了ScheduledExecutorService及Iterator,一个基于netty事件循环处理机制而实现的线程池抽象,遍历返回为其子类EventExecutor(一个事件处理器)

|EventLoopGroup接口 ：继承了EventExecutorGroup，提供了Channel的注册，且重写遍历方法next()的返回值为EventLoop

EventLoop接口：继承了EventLoopGroup接口，用于处理Channel连接生命周期中所发生的事件。当Channel在一个EventLoop上注册了，EventLoop会负责处理该Channel上的所有I/O操作，一般一个EventLoop实例负责处理多个Channel。

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

一个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数据帧封装的基类，其提供了以下实现类：

为了简单处理SimpleChannelInboundHandler消息类型选取了文本帧TextWebSocketFrame，该类的text()方法会以UTF-8的字符串格式获取WebSocket的传输内容，每次server与client的WebSocket交互都以字符串传输即可。

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

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

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

@Component@ChannelHandler.Sharable@Slf4j@AllArgsConstructorpublic 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>的部分方法源码如下：

@Overridepublic 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;

@Slf4j@SuppressWarnings({"rawtypes", "unchecked"})@Componentpublic 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);    }}

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);}

@Componentpublic 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@Configurationpublic 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");    }}

@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。

效果演示私聊演示群聊演示结语

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

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

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

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

附

代码地址：github：Wilson-He/netty-simple-chat

如果对个人策略模式的演变与实现思路感兴趣的可以看我上篇文章：Spring+策略模式=无策略?

(1个月前写的代码跟一个月后写的代码区别是很大的，就算别人问我一周前写的代码我都会一脸懵逼，所以不要问我以前代码的细节问题)