【如何构建商业级别聊天系统】开始搭建你的聊天服务 Netty 最佳实践快速入门

以你之姓@ 2022-09-11 03:17 136阅读 0赞

#### Netty 快速入门最佳实践 ####

官网地址 ：https://netty.io/wiki/user-guide-for-4.x.html

特关人上人 [dying 搁浅神秘连接点击获取知识密码][dying]

#### 写在前面 ####

本文基于 Spring Boot 框架环境进行，读者可自行搭建环境，或者自行调整部分代码以适配自己的环境。

同时你需要有一定的基础概念了解，推荐查看我的其他关于 Netty 的文章。文中涉及到的一些概念只会简单叙述一下，不做展开讲解。  
[Netty 架构概述][Netty]

#### 编写你的 Netty 服务 ####

首先你需要构建一个 Netty 服务，来暴露你的通信地址，这里我最终会构建一个 简单的 WebSocket 服务，所以这里取名叫 WebSocketServer

这里通过实现 `ApplicationListener<ContextRefreshedEvent>`可以使得 Spring iOC 容器在初始化 bean 后启动 我们的 Netty 服务

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.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;
    
    /** * @author: wangzibin **/
    @Slf4j
    @Component
    public class WebsocketServer implements ApplicationListener<ContextRefreshedEvent> { 
    
        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;
    
        @Override
        public void onApplicationEvent(ContextRefreshedEvent event) { 
            log.info("Starting WebSocketServer on port {} ", WEBSOCKET_PORT);
    
            long startMillis = System.currentTimeMillis();
          	// boss 线程 1 个用于处理绑定连接到 worker
            bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("WebsocketBoss"));
          	// worker 线程 2 倍 cpu 核数 用于处理连接的数据传输
            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();
                  					// 这里 pipline 是添加自定义处理逻辑的地方
                            pipeline.addLast(new 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 服务来说，最简单的不是 “Hello Word” ，而是什么都不做，首先编写一个消息拒绝处理服务

import io.netty.buffer.ByteBuf;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /** * 拒绝服务, 用于基础测试 */
    public class DiscardServerHandler extends ChannelInboundHandlerAdapter { 
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {  
          	// 什么都不管直接丢弃
            ((ByteBuf) msg).release();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            // 该方法将在 Netty I/O 错误，或者实现的 Handler 抛出异常时进行回调，通常你可以在这里写一些异常的记录或者相关的处理 
            cause.printStackTrace();
            ctx.close();
        }
    }

ChannelPipeline pipeline 注意这个我们添加自定义 Handler 的管道，其在设计上分两个方向，数据入向 Inbound 和 数据出向 outbound，这个是个相对的概念，通常来说我们想处理出入的消息数据直接继承或者实现 Netty 为我们提供的接口或者类就好了，这里我们选择的是继承 `ChannelInboundHandlerAdapter`并重写他的方法。

[`ByteBuf`][ByteBuf]是一个引用计数对象，必须通过`release()`方法显式释放。请记住，释放传递给处理程序的任何引用计数对象是处理程序的责任。通常，`channelRead()`处理程序方法的实现方式如下：

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) { 
        try { 
            // Do something with msg
        } finally { 
            ReferenceCountUtil.release(msg);
        }
    }

这个时候服务的消息以及可以在你的控制下运行了，但是很显然，我们拒绝了所有的消息，所以这个服务什么都做不了，让我们做个改版。修改 channelRead 方法将你收到的数据打印出来。

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) { 
        ByteBuf in = (ByteBuf) msg;
        try { 
            while (in.isReadable()) {  // (1)
                System.out.print((char) in.readByte());
                System.out.flush();
            }
        } finally { 
            ReferenceCountUtil.release(msg); // (2)
        }
    }

#### 编写客户端服务 ####

ok ，看起来我们的服务已经准备就绪了，那如何验证服务是否如我们所愿正常运行呢，这里我们编辑客户端服务来进行数据的发送。逻辑上同服务端是近似的，但会有细微的差别。

import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.util.CharsetUtil;
    import io.netty.util.concurrent.EventExecutorGroup;
    
    /** * @author: wangzibin **/
    public class NettyClient { 
        static class NettyClientHandler extends ChannelInboundHandlerAdapter { 
            /** * 当客户端连接服务器完成就会触发该方法 * * @param ctx * @throws Exception */
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception { 
                for (int i=0;i<10;i++){ 
                    ByteBuf buf = Unpooled.copiedBuffer("HelloServer", CharsetUtil.UTF_8);
                  	// 这里直接将 “HelloServer” 发送给服务端
                    ctx.writeAndFlush(buf);
                }
    
            }
    
            //当通道有读取事件时会触发，即服务端发送数据给客户端
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
                ByteBuf buf = (ByteBuf) msg;
                System.out.println("收到服务端的消息:" + buf.toString(CharsetUtil.UTF_8));
                System.out.println("服务端的地址： " + ctx.channel().remoteAddress());
            }
    
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
                cause.printStackTrace();
                ctx.close();
            }
        }
    
        public static void main(String[] args) throws Exception { 
            //客户端需要一个事件循环组
            EventLoopGroup group = new NioEventLoopGroup();
            try { 
                //创建客户端启动对象
                //注意客户端使用的不是 ServerBootstrap 而是 Bootstrap
                Bootstrap bootstrap = new Bootstrap();
                //设置相关参数
                bootstrap.group(group) //设置线程组
                        .channel(NioSocketChannel.class) // 使用 NioSocketChannel 作为客户端的通道实现
                        .handler(new ChannelInitializer<SocketChannel>() { 
    
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception { 
                                //加入处理器
                                channel.pipeline().addLast(new NettyClientHandler());
                            }
                        });
                System.out.println("netty client start");
                //启动客户端去连接服务器端
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 18680).sync();
                //对关闭通道进行监听
                channelFuture.channel().closeFuture().sync();
            } finally { 
                group.shutdownGracefully();
            }
        }
    }

此时，运行服务端后再运行客户端，服务端已经可以在控制台打印出客户端发送来的数据 “HellowServer” 了，但显然有个问题，基于 TCP 的流式传输所引发的粘包现象，使得我们接受到的 HellowServer 粘连到一起了。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16][]

当然，解决这种粘包拆包的方法有很多种，可以根据你的业务场景自行选择，比如长度标志位等。通常我们会编写相关的 【**编解码器**】 这里我们做一个简单的处理示范一下。

#### 升级你的服务添加编解码器 ####

Netty 内置了很多开箱即用的编解码器，方便我们直接继承其特性，使我们可以将更多的精力放到业务逻辑上。

添加 HelloServerReplayingDecoder 解码器，这里我们选择直接继承 ReplayingDecoder 来使用其特性

import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    /** * @author: wangzibin **/
    public class HelloServerReplayingDecoder extends ReplayingDecoder<Void> { 
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
            out.add(in.readBytes(11));
        }
    }

简单看下作者对 ReplayingDecoder 的解释

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16 1]

简单来说，该 Handler 会等待 11 个字节才传入下一个 Handler 中进行相关的处理。这样就有效的将客户端发送的 HelloServer 分隔开来

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16 2]

别忘了将解码器添加到你服务端的 pipline 中 `pipeline.addLast(new HelloServerReplayingDecoder());`

将数据写回给客户端，让我们重新完善下 `DiscardServerHandler`

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    import io.netty.util.ReferenceCountUtil;
    import lombok.extern.slf4j.Slf4j;
    
    
    /** * @description: 拒绝服务, 用于基础测试 * @author: wangzibin **/
    @Slf4j
    public class DiscardServerHandler extends ChannelInboundHandlerAdapter { 
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            // Discard the received data silently.
            ByteBuf in = (ByteBuf) msg;
            StringBuilder stringBuilder = new StringBuilder();
            while (in.isReadable()) { 
                    char readByte = (char) in.readByte();
                    stringBuilder.append(readByte);
            }
            System.out.println(stringBuilder.toString());
            ByteBuf buf = Unpooled.copiedBuffer(stringBuilder.toString()+" return ", CharsetUtil.UTF_8);
            ctx.writeAndFlush(buf);
    
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            // Close the connection when an exception is raised.
            cause.printStackTrace();
            ctx.close();
        }
    }

到此为止，一个完整的数据交换流程就完成了，相信大家也看到有个明显的问题，我们使用 ByteBuf 收发数据有诸多不便，通常来说我们会选择其他更加方便高效的传输方式，如谷歌的 protobuf，同时在编解码的阶段将数据流直接转换成我们约定好的 类型实体进行统一处理。

#### 替换 ByteBuf 为 POJO ####

新建一个类型 UserMessage

import lombok.AllArgsConstructor;
    import lombok.Builder;
    import lombok.Data;
    
    import java.io.Serializable;
    
    /** * @author: wangzibin **/
    @Data
    @Builder
    @AllArgsConstructor
    public class UserMessage implements Serializable { 
        private int id;
    }

重写解码器

import com.star.im.websocket.handler.demo.message.UserMessage;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    /** * @author: wangzibin **/
    public class HelloServerReplayingDecoder extends ReplayingDecoder<Void> { 
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
            int age = in.readInt();
            out.add(new UserMessage(age));
        }
    }

优化“拒绝服务逻辑”

import com.star.im.websocket.handler.demo.message.UserMessage;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    import io.netty.util.ReferenceCountUtil;
    import lombok.extern.slf4j.Slf4j;
    
    
    /** * @description: 拒绝服务, 用于基础测试 **/
    @Slf4j
    public class DiscardServerHandler extends ChannelInboundHandlerAdapter { 
    
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            // Discard the received data silently.
            log.info("channelRead ctx={}, msg={}", ctx, msg);
            if (msg instanceof String){ 
                System.out.println(msg);
                ctx.writeAndFlush(msg);
            }else if (msg instanceof UserMessage){ 
                UserMessage userMessage= (UserMessage) msg;
                System.out.println(userMessage);
            }else { 
                ByteBuf in = (ByteBuf) msg;
                StringBuilder stringBuilder = new StringBuilder();
                while (in.isReadable()) { 
                    char readByte = (char) in.readByte();
                    stringBuilder.append(readByte);
                }
                System.out.println(stringBuilder.toString());
                ByteBuf buf = Unpooled.copiedBuffer(stringBuilder.toString()+" return ", CharsetUtil.UTF_8);
                ctx.writeAndFlush(buf);
            }
    
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            // Close the connection when an exception is raised.
            cause.printStackTrace();
            ctx.close();
        }
    }

替换客户端发送逻辑代码

@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
        for (int i = 0; i < 30; i++) { 
            ByteBuf buf = Unpooled.copyInt(i);
            ctx.writeAndFlush(buf);
        }
    }

这个时候我们就可以直接操作定义好的类型了

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_14_color_FFFFFF_t_70_g_se_x_16]

同样的，我们也可以如法炮制编码器，这样可以在双端实现直接传递类型数据，那么真正的解析逻辑会在最初的编解码阶段进行转换，后续我们只需要专注于业务逻辑即可。

#### 写在最后 ####

本篇文章带你简单编写了一个 Netty 服务，相信你已经有所入门。但是别着急，商业级别聊天系统的构建才刚刚开始。

下篇文章，我们将升级 Demo 项目为一个正式的 WebSocket 服务，实现网页端与服务器的通信。

我是 dying 搁浅，我们的路才刚刚开始。

there Life is Strange

[dying]: https://mp.weixin.qq.com/s?__biz=Mzg2NzA1MjY1OA==&tempkey=MTEzMF9BZDJPbXFiSnFHeFMrR1BQWXNGNTUteGNkNGdyMlRkYXp6S3I2aXV2Tk96cDgyOXlFeV84dVE4cDAzbWJwbFRVM3VwMWlfUmtlVjVwN1RCTjJVMXk2X3BiQ3doNWJQZERwVHE4TVVnODBZckJ6VTE4YkJ1Mnd1dnlhYUE3VHRUVUE3MU9hNm9oNDZtTkdvVFFqQ1lRdGRySEQ0cmk2NmgwYURkZDh3fn4=&chksm=4e403f947937b6829b9add7c04dc107db522d2a5c4804172993e0f48e234c4e99bb2536eeddd&scene=0&xtrack=1&previewkey=%252BK97yR3vsbgNCKfj9vx%252FpsNS9bJajjJKzz%252F0By7ITJA%253D#wechat_redirect
[Netty]: https://blog.csdn.net/w903328615/article/details/119680478
[ByteBuf]: https://netty.io/4.1/api/io/netty/buffer/ByteBuf.html
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/3baf6cc6d6eb473a954ee2b845cedb90.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/64bfc8bb658d4f6ea97a44a2eab4ca14.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/26d7b511fc7b49a38daa685c5215e833.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_14_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/165c7946c7e845b5859730a1c632de7d.png