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

傷城~ 2022-09-14 00:16 164阅读 0赞

### 写在前面 ###

承接上文：[【如何构建商业级别聊天系统】 Netty 构建 WebSocket 服务 上篇][Netty _ WebSocket _]

上两张效果图吧：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16 1]

在构建 WebSocket 服务的上篇，我们介绍了几个核心的由 Netty 提供的处理 Http 以及升级 WebScoket 协议的 Handler

那么在建立好连接后后续的数据传输和处理就需要我们来编写自己的 Handler 来进行处理了，无论你是想讲消息入库还是分发都可以通过定制 Handler 并将其加入到 pipline 中来实现。

这里贴上上篇的代码方便大家查看

/** * @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);
        }
    }

### 自定义 Handler 处理连接和分发 ###

pipeline.addLast(commonConnectHandler);
    pipeline.addLast(discardServerHandler);

本篇的重点在于上面这两个 Handler 的处理逻辑。

此时我们先实现一个简单的需求：在多个客户端连接上的 WebSocket 服务后可以进行消息的传递，将发送的消息发送给所有在线的客户端。相当于一个简单的聊天室

顺着这个需求来分析一下如何实现：

1.  首先服务端需要保存所有的连接以便消息来临后可以进行广播发送。
2.  我们需要一个对消息进行处理的 Handler ，进行消息分发。

这个时候职责就很清晰了

**commonConnectHandler 负责连接管理**

**discardServerHandler 负责数据分发** （当然这个名称可能起的不是很对，目前还是 Demo 就先这样吧哈哈）

#### CommonConnectHandler 连接管理 ####

import com.star.im.websocket.mgr.ConnectionManager;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    /** * @program: star-im * @description: 连接管理 **/
    @Component
    @ChannelHandler.Sharable
    public class CommonConnectHandler extends ChannelInboundHandlerAdapter { 
    
        @Resource
        private ConnectionManager connectionManager;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    				// 注册所有连接
            connectionManager.registerConnection(ctx.channel());
            ctx.fireChannelRead(msg);
        }
    }

import io.netty.channel.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import java.util.*;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.atomic.AtomicInteger;
    @Slf4j
    @Component
    public class ConnectionManager { 
    
        private final ConcurrentHashMap<Channel, NettyConnection> channelMap = new ConcurrentHashMap<>(10000);
        private final Set<NettyConnection> connectionPool = Collections.synchronizedSet(new HashSet<>(10000));
    
        private final static AtomicInteger ATOMIC_INTEGER = new AtomicInteger();
    
        public boolean registerConnection(Channel conn) { 
            if (channelMap.containsKey(conn)) { 
                return true;
            }
            int num = ATOMIC_INTEGER.incrementAndGet();
            NettyConnection nettyConnection = NettyConnection.builder().channel(conn).userName(num + " 号用户").cid(num).build();
            connectionPool.add(nettyConnection);
            channelMap.put(conn, nettyConnection);
            log.info("registerConnection {}", nettyConnection);
            return true;
        }
    
        public void removeConnection(NettyConnection channel) { 
            connectionPool.remove(channel);
            channelMap.remove(channel.getChannel());
            destroyConnection(channel);
        }
    
        private void destroyConnection(NettyConnection conn) { 
            log.info("destroyConnection {}", conn);
            conn.getChannel().close();
    
        }
    
    
        public NettyConnection getConnection(Channel channel) { 
            return channelMap.get(channel);
        }
    
        public Set<NettyConnection> getConnectionPool() { 
            return connectionPool;
        }
    
    
    }

import io.netty.channel.Channel;
    import lombok.*;
    
    import java.io.Serializable;
    
    @ToString(callSuper = true)
    @EqualsAndHashCode()
    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public class NettyConnection implements Serializable { 
    
        private long cid;
    
        private String userName;
    
        private transient Channel channel;
    
    }

对于连接管理，核心类主要为以上几个

NettyConnection 为存储连接的实体类。

ConnectionManager 为连接管理类用于管理生成 NettyConnection 。其中最重要的就是 Channel 这个类，最终的数据发送都是通过改类完成，同时这里简单地生成了一下 cid 来标识用户。

CommonConnectHandler 的作用就是处理每个进来的连接，并将它们注册到 ConnectionManager 中。

#### DiscardServerHandler 数据分发 ####

import com.star.im.websocket.mgr.ConnectionManager;
    import com.star.im.websocket.mgr.NettyConnection;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.util.CharsetUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.Objects;
    import java.util.Set;
    
    @Slf4j
    @Component
    @ChannelHandler.Sharable
    public class DiscardServerHandler extends ChannelInboundHandlerAdapter { 
    
        @Resource
        private ConnectionManager connectionManager;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            if (msg instanceof String) { 
                log.info("channelRead String, msg={}", msg);
                System.out.println(msg);
                String msgStr = String.valueOf(msg);
    
                Set<NettyConnection> channelList = connectionManager.getConnectionPool();
                Channel currentChannel = ctx.channel();
                NettyConnection currentConnection = connectionManager.getConnection(currentChannel);
                String userName = currentConnection.getUserName();
                for (NettyConnection connection : channelList) { 
    
                    String finalMsg = Objects.equals(connection, currentConnection) ? "你说:" + msgStr : userName + ":" + msgStr;
                    ByteBuf buf = Unpooled.copiedBuffer(finalMsg, CharsetUtil.UTF_8);
                    ChannelFuture channelFuture = connection.getChannel().writeAndFlush(buf);
                    channelFuture.addListener((ChannelFutureListener) future -> { 
                        if (!future.isSuccess()) { 
                            log.error(">>> String failure! ch=" + connection + ", finalMsg=" + finalMsg,
                                    future.cause());
                            connectionManager.removeConnection(connection);
                        }
                    });
                }
    
            } else { 
                log.info("channelRead ctx={}, msg={}", ctx, msg);
                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();
        }
    }

这里主要的处理逻辑在第一块 if 当中，在本案例中我们主要处理 String 类型的消息，通过 ConnectionManager 获取所有在线的连接，在分发的时候文案区分自己和他人。

#### H5 页面 ####

以上就是服务端的内容了，这里给出一个测试用的 H5 页面，方便测试效果

<!doctype html>
    <html lang="zh-CN">
    
    <head>
       
       <meta charset="utf-8">
       <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
    
       
       <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@4.6.0/dist/css/bootstrap.min.css">
    
       <title>Hello, world!</title>
    </head>
    
    <body>
       <div id="app" style="margin-top: 50px;">
    
    
    
          <div class="container">
             <div class="row">
    
                <div class="col-sm">
                   <h3>最后一条消息：{
       {newMessage}}</h3>
                   <div>
                      <div class="form-group">
                         <label for="exampleInputEmail1">消息</label>
                         <input v-model="message" type="text" class="form-control" id="exampleInputEmail1" aria-describedby="emailHelp" @keyup.enter="sendMsg">
                         <small id="emailHelp" class="form-text text-muted">说点什么</small>
                      </div>
                      <button v-on:click="sendMsg" class="btn btn-primary">发送</button>
                   </div>
                </div>
                <div class="col-sm">
                   <div class="jumbotron">
                      <p class="lead" v-for="item in msgList">{
       {item}}</p>
            
                    </div>
                </div>
               
             </div>
          </div>
    
    
    
    
    
       </div>
    
    </body>
    
    <script src="https://cdn.jsdelivr.net/npm/vue@2/dist/vue.js"></script>
    <script type="text/javascript"> // 打开一个 web socket var ws = new WebSocket("ws://localhost:18680/chatws"); var app = new Vue({ el: '#app', data: { message: 'Hello Vue!', newMessage:"", msgList: [""] }, created: function () { var that=this; ws.onopen = function () { // Web Socket 已连接上，使用 send() 方法发送数据 ws.send("我来了"); console.log("数据已发送：我来了"); ws.onmessage = function (evt) { var received_msg = evt.data; const f1 = received_msg.text(); console.log("数据已接收 evt.data：", f1) f1.then(a => { // 变量a的作用域只能在回调内。在外部声明时，因为线程的争夺当前变量的值是不可靠的 console.log(a) that.newMessage=a; that.msgList.push(a); }) }; }; }, // 在 `methods` 对象中定义方法 methods: { sendMsg: function (event) { if(!this.message){ return; } ws.send(this.message); this.message=""; }, } }) </script>
    
    
    <script src="https://cdn.jsdelivr.net/npm/jquery@3.5.1/dist/jquery.slim.min.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/bootstrap@4.6.0/dist/js/bootstrap.bundle.min.js"></script>
    
    </html>

#### 最后 ####

本篇向大家介绍了自己定制的 Handler 的处理机制，以及通常的连接管理和分发逻辑，整体具备一定的参考意义，后续将以此为基础进行优化迭代。

[Netty _ WebSocket _]: https://blog.csdn.net/w903328615/article/details/120444151
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/8a3400408cd741c3a78e27f38a7b2893.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAZHlpbmcg5pCB5rWF_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/298ace255f5e4c44aaf6dfd950623495.png