基于Netty实现websocket集群部署实现方案

不念不忘少年蓝@ 2022-11-16 10:57 980阅读 0赞

# 基于Netty实现websocket集群部署实现方案 #

每天多学一点点~  
话不多说，这就开始吧…

### 文章目录 ###

*  基于Netty实现websocket集群部署实现方案
 *   *  1.前言
     *  2. 整体思路
     *  3. 代码demo
     *  4. 测试
     *  5.结语

## 1.前言 ##

最近公司在做saas平台，其中涉及到重构一个无人机项目。无人机推流拉流用了腾讯云直播、点播功能。安卓端集成了大疆的sdk，需要在飞无人机的时候一直推送飞行信息(比如飞行高度，飞行路线、风向什么的)。  
之前用的tomcat自带的websocket,spring-boot-starter-websocke集成，但是性能可能有点问题。这次重构，打算换成netty。因为是saas化服务，领导还要求可集群部署，这里提供一种解决方案----redis的pub、sub（当然大型项目最好还是zk）。

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

## 2. 整体思路 ##

1.  跨服务之间案例采用redis的发布和订阅进行传递消息。多个实例监听同一个channel。客户端上线存入redis，下线清除redis。
2.  用户A在发送消息给用户B时候，需要传递B的channeId，以用于服务端进行查找channeId所属是否自己的服务内。
3.  若本次channel在A实例的本地缓存中能找到，说明属于自己服务，直接发送。若不在，则通过redis的pub到其他服务实例，其他实例监听到消息之后，判断这个channel是否在本地，在就发送。

当然，redis 的pub sub本身也是有缺陷的，比如： 数据可靠性无法保证、扩展性差、资源消耗高，如果是正式比较大型的可以换成zk，思路差不多。

## 3. 代码demo ##

nettywebsocket
    ├─ main
    │ 	├─ java
    │ 	│ 	└─ com
    │ 	│ 		└─ example
    │ 	│ 			└─ demo
    │ 	│ 				├─ advice
    │ 	│ 				│ 	├─ AdviceController.java
    │ 	│ 				│ 	├─ BusinessException.java
    │ 	│ 				│ 	├─ ErrorConstant.java
    │ 	│ 				│ 	└─ NotFoundController.java
    │ 	│ 				├─ domain
    │ 	│ 				│ 	├─ CommonResult.java
    │ 	│ 				│ 	├─ IErrorCode.java
    │ 	│ 				│ 	├─ MyMessage.java
    │ 	│ 				│ 	├─ ResultCode.java
    │ 	│ 				│ 	├─ ServerInfo.java
    │ 	│ 				│ 	└─ UserChannelInfo.java
    │ 	│ 				├─ NettywebsocketApplication.java
    │ 	│ 				├─ redis
    │ 	│ 				│ 	├─ config
    │ 	│ 				│ 	│ 	└─ RedisConfig.java
    │ 	│ 				│ 	├─ RedisChannelListener.java
    │ 	│ 				│ 	└─ RedisUtil.java
    │ 	│ 				├─ server
    │ 	│ 				│ 	├─ HttpRequestHandler.java
    │ 	│ 				│ 	├─ NioWebSocketChannelInitializer.java
    │ 	│ 				│ 	├─ WebsocketServer.java
    │ 	│ 				│ 	└─ WebSocketServerHandler.java
    │ 	│ 				├─ util
    │ 	│ 				│ 	├─ CacheUtil.java
    │ 	│ 				│ 	├─ DirectoryTreeV1.java 
    │ 	│ 				│ 	├─ MsgUtil.java
    │ 	│ 				│ 	└─ NetUtil.java
    │ 	│ 				└─ web
    │ 	│ 					└─ WebSocketController.java
    │ 	└─ resources
    │ 		├─ application.yml
    │ 		└─ static
    │ 			├─ index.html
    │ 			├─ index2.html
    │ 			└─ index3.html
    └─ test
    	└─ java
    		└─ com
    			└─ example
    				└─ demo
    					└─ NettywebsocketApplicationTests.java

这里列出一些关键代码  
**实体类**

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 0:38 * @description: TODO 定义信息传输协议，这个看似简单但非常重要，每一个通信的根本就是定义传输协议信息 * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    @Data
    @AllArgsConstructor
    public class MyMessage { 
    
        //发送给某人，某人channelId
        private String toChannelId;
    
        //消息内容
        private String content;
    }
    服务端信息
    @Data
    @AllArgsConstructor
    public class ServerInfo { 
    
        //IP
        private String ip;
    
        //端口
        private int port;
    
        //启动时间
        private Date openDate;
    }
    
    # 用户管道信息；记录某个用户分配到某个服务端
    @Data
    @AllArgsConstructor
    public class UserChannelInfo { 
    
        //服务端：IP
        private String ip;
    
        //服务端：port
        private int port;
    
        //channelId
        private String channelId;
    
        //链接时间
        private Date linkDate;
    
    }

**RedisConfig.java** 配置redis 序列化 以及 监听

**RedisChannelListener.java** 监听业务逻辑

**RedisUtil.java** redis工具类

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 0:42 * @description: TODO redis工具类 * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    @Component
    public class RedisUtil { 
    
        @Autowired
        private StringRedisTemplate redisTemplate;
    
        public void pushObj(UserChannelInfo userChannelInfo) { 
            redisTemplate.opsForHash().put("uav-flight-user",
                    userChannelInfo.getChannelId(), JSON.toJSONString(userChannelInfo));
        }
    
        public List<UserChannelInfo> popList() { 
            List<Object> values = redisTemplate.opsForHash().values("uav-flight-user");
            if (null == values) { 
                return new ArrayList<>();
            }
    
            List<UserChannelInfo> userChannelInfoList = new ArrayList<>();
    
            for (Object strJson : values) { 
                userChannelInfoList.add(JSON.parseObject(strJson.toString(), UserChannelInfo.class));
            }
            return userChannelInfoList;
        }
    
        public void remove(String channelId) { 
            redisTemplate.opsForHash().delete("uav-flight-user", channelId);
        }
    
        public void clear() { 
            redisTemplate.delete("uav-flight-user");
        }
    
    
        public void push(String channel, String message) { 
            redisTemplate.convertAndSend(channel, message);
    
        }
    
    }

**CacheUtil.java** 缓存必要信息，用于业务流程处理

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 0:38 * @description: TODO * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    public class CacheUtil { 
    
        // 缓存channel
        public static Map<String, Channel> cacheChannel = Collections.synchronizedMap(new HashMap<String, Channel>());
    
        // 缓存服务信息
        public static Map<Integer, ServerInfo> serverInfoMap = Collections.synchronizedMap(new HashMap<Integer, ServerInfo>());
    
        // 缓存 websocket 服务端
        public static Map<Integer, WebsocketServer> serverMap = Collections.synchronizedMap(new HashMap<Integer, WebsocketServer>());
    
    }

**WebsocketServer.java** 基于netty的websocket服务端

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 0:42 * @description: TODO * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    @Slf4j
    @Service
    public class WebsocketServer implements Callable<Channel> { 
    
    
        @Value("${netty.port}")
        private int port;
    
        //配置服务端NIO线程组
        private final EventLoopGroup bossGroup = new NioEventLoopGroup();
    
        private final EventLoopGroup workerGroup = new NioEventLoopGroup();
    
    
        @Autowired
        private NioWebSocketChannelInitializer nioWebSocketChannelInitializer;
    
        private Channel channel = null;
    
        public Channel getChannel() { 
            return this.channel;
        }
    
        public void setChannel(Channel channel) { 
            this.channel = channel;
        }
    
        /** * 这里有点问题，换句话，应该直接 在系统启动的时候初始化一次更好，可以优化下 * @return * @throws Exception */
        @Override
        public Channel call() throws Exception { 
            ChannelFuture channelFuture = null;
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            try { 
                //boss辅助客户端的tcp连接请求 worker负责与客户端之前的读写操作
                serverBootstrap.group(bossGroup, workerGroup)
                        //配置客户端的channel类型
                        .channel(NioServerSocketChannel.class)
                        //配置TCP参数，握手字符串长度设置
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        //TCP_NODELAY算法，尽可能发送大块数据，减少充斥的小块数据
                        .option(ChannelOption.TCP_NODELAY, true)
                        //开启心跳包活机制，就是客户端、服务端建立连接处于ESTABLISHED状态，超过2小时没有交流，机制会被启动
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        //配置固定长度接收缓存区分配器
                        .childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(592048))
                        // websocket 初始化 handler
                        .childHandler(nioWebSocketChannelInitializer);
    
                log.info("Netty Websocket服务器启动完成，已绑定端口 " + port + " 阻塞式等候客户端连接");
    
                channelFuture = serverBootstrap.bind(port).sync();
                this.channel = channelFuture.channel();
                // 因为是通过接口调用所以这里不能异步
    // channel.closeFuture().sync();
            } catch (Exception e) { 
                e.printStackTrace();
                throw new RuntimeException(e.getMessage());
            }
    
            return channel;
        }
    
    
        /** * 摧毁 */
        public void destroy() { 
            if (null == channel) { 
                return;
            }
            channel.close();
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    
    }

**NioWebSocketChannelInitializer.java** 初始化channel

@Service
    public class NioWebSocketChannelInitializer extends ChannelInitializer<SocketChannel> { 
    
        @Autowired
        private WebSocketServerHandler webSocketServerHandler;
    
        @Autowired
        private HttpRequestHandler httpRequestHandler;
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception { 
            //设置log监听器，并且日志级别为debug，方便观察运行流程
            ch.pipeline().addLast("logging", new LoggingHandler("DEBUG"));
            //设置解码器
            ch.pipeline().addLast("http-codec", new HttpServerCodec());
            //聚合器，使用websocket会用到 把HTTP头、HTTP体拼成完整的HTTP请求
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
            //用于大数据的分区传输
            ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
            // 用于http 升级成 websocket
            ch.pipeline().addLast("http-handler", httpRequestHandler);
            //自定义的业务handler 处理websocket
            ch.pipeline().addLast("handler", webSocketServerHandler);
        }
    }

**HttpRequestHandler.java** 用于对http 协议 升级 的 handler

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 1:45 * @description: TODO 用于对http 协议 升级 的 handler * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    @Component
    @Sharable
    public class HttpRequestHandler extends SimpleChannelInboundHandler<Object> { 
    
        /** * 读取 数据 * 描述：读取完连接的消息后，对消息进行处理。 * 这里仅处理HTTP请求，WebSocket请求交给下一个处理器。 */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { 
            if (msg instanceof FullHttpRequest) { 
                handleHttpRequest(ctx, (FullHttpRequest) msg);
            } else if (msg instanceof WebSocketFrame) { 
                ctx.fireChannelRead(((WebSocketFrame) msg).retain());
            } 
        }
    
        /** * 描述：处理Http请求，主要是完成HTTP协议到Websocket协议的升级 * @param ctx * @param req */
        private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { 
            if (!req.decoderResult().isSuccess()) { 
                sendHttpResponse(ctx, req,
                        new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
                return;
            }
    
            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                    "ws:/" + ctx.channel() + "/websocket", null, false);
            WebSocketServerHandshaker handshaker = wsFactory.newHandshaker(req);
    
            if (handshaker == null) { 
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else { 
                handshaker.handshake(ctx.channel(), req);
            }
        }
        
        private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) { 
            // 返回应答给客户端
            if (res.status().code() != 200) { 
                ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
                res.content().writeBytes(buf);
                buf.release();
            }
            // 如果是非Keep-Alive，关闭连接
            boolean keepAlive = HttpUtil.isKeepAlive(req);
            ChannelFuture f = ctx.channel().writeAndFlush(res);
            if (!keepAlive) { 
                f.addListener(ChannelFutureListener.CLOSE);
            }
        }
    
        /** * 描述：异常处理，关闭channel */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
    }

**WebSocketServerHandler.java** 处理接收消息，上线下线等业务逻辑handler

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 1:37 * @description: TODO * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    @Slf4j
    @Service
    @ChannelHandler.Sharable       // 因是通过注入的方式，而不是通过new,共享channel
    public class WebSocketServerHandler extends SimpleChannelInboundHandler<WebSocketFrame> { 
    
        @Autowired
        private RedisUtil redisUtil;
    
        private WebSocketServerHandshaker handshaker;
    
        /** * 读取 消息 * * @param ctx * @param frame * @throws Exception */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception { 
    
            // 关闭请求
            if (frame instanceof CloseWebSocketFrame) { 
                handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
                return;
            }
            // ping请求
            if (frame instanceof PingWebSocketFrame) { 
                ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
                return;
            }
            // 只支持文本格式，不支持二进制消息
            if (!(frame instanceof TextWebSocketFrame)) { 
                sendErrorMessage(ctx, "仅支持文本(Text)格式，不支持二进制消息");
            }
    
            // 客服端发送过来的消息
            String request = ((TextWebSocketFrame) frame).text();
            log.info("服务端收到新信息：" + request);
            try { 
                //接收msg消息{此处已经不需要自己进行解码}
                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 接收到消息内容：" + request);
    
                MyMessage msgAgreement = MsgUtil.json2Obj(request.toString());
    
                String toChannelId = msgAgreement.getToChannelId();
                //判断接收消息用户是否在本服务端
                Channel channel = CacheUtil.cacheChannel.get(toChannelId);
                if (null != channel) { 
                    channel.writeAndFlush(new TextWebSocketFrame(MsgUtil.obj2Json(msgAgreement) + " lalalalalalalalalalalalal"));
                    return;
                }
    
                //如果为NULL则接收消息的用户不在本服务端，需要push消息给全局
                log.info("接收消息的用户不在本服务端，PUSH！");
                redisUtil.push("uav-flight-message", MsgUtil.obj2Json(msgAgreement));
    
    
            } catch (Exception e) { 
                sendErrorMessage(ctx, "JSON字符串转换出错！");
                e.printStackTrace();
            }
        }
    
        /** * 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接 */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            ctx.close();
            // 1. 清除 redis
            redisUtil.remove(ctx.channel().id().toString());
            // 2. 清除缓存
            CacheUtil.cacheChannel.remove(ctx.channel().id().toString(), ctx.channel());
            System.out.println("异常信息：\r\n" + cause.getMessage());
        }
    
        /** * 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据 * * @param ctx * @throws Exception */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            SocketChannel channel = (SocketChannel) ctx.channel();
            System.out.println("客户端上线");
            System.out.println("客户端信息：有一客户端链接到本服务端。channelId：" + channel.id());
            System.out.println("客户端IP:" + channel.localAddress().getHostString());
            System.out.println("客户端Port:" + channel.localAddress().getPort());
            System.out.println("客户端信息完毕");
    
            //保存用户信息
            UserChannelInfo userChannelInfo = new UserChannelInfo(channel.localAddress().getHostString(),
                    channel.localAddress().getPort(), channel.id().toString(), new Date());
            // 放入 redis 和 缓存
            redisUtil.pushObj(userChannelInfo);
            CacheUtil.cacheChannel.put(channel.id().toString(), channel);
            //通知客户端链接建立成功
            String str = "通知客户端链接建立成功" + " " + new Date() + " " + channel.localAddress().getHostString() + "\r\n";
            ctx.writeAndFlush(MsgUtil.buildMsg(channel.id().toString(), str));
        }
    
    
        /** * 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据 * * @param ctx * @throws Exception */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
            System.out.println("客户端断开链接" + ctx.channel().localAddress().toString());
            // 移除 redis 和 缓存
            redisUtil.remove(ctx.channel().id().toString());
            CacheUtil.cacheChannel.remove(ctx.channel().id().toString(), ctx.channel());
        }
    
    
        private void sendErrorMessage(ChannelHandlerContext ctx, String errorMsg) { 
            String responseJson = "不支持二进制消息";
            ctx.channel().writeAndFlush(new TextWebSocketFrame(responseJson));
        }
    
    }

**WebSocketController.java** 控制层，开启、关闭netty，查看服务端、用户管道列表(其实这里可以根据业务逻辑，项目启动时候就启动netty，也不一定非要调接口开启)

/* ━━━━━━佛祖保佑━━━━━━ * ,;,,; * ,;;'( 社 * __ ,;;' ' \ 会 * /' '\'~~'~' \ /'\.) 主 * ,;( ) / |. 义 *,;' \ /-.,,( ) \ 码 * ) / ) / )| 农 * || || \) * (_\ (_\ * ━━━━━━永无BUG━━━━━━ * @author ：zjq * @date ：2021/4/7 1:20 * @description: TODO http://www.websocket-test.com 在线测试 * {"content":"xxxx","toChannelId":"db3abfed"} 测试数据 toChannelId 在 redis 中 * @version: V1.0 * @slogan: 天下风云出我辈，一入代码岁月催 */
    @RestController
    @RequestMapping
    @Slf4j
    public class WebSocketController { 
    
    
        @Value("${netty.port}")
        private int port;
    
        @Autowired
        private RedisUtil redisUtil;
    
        @Autowired
        private WebsocketServer websocketServer;
    
        //默认线程池 先意思意思 这么用
        private static ExecutorService executorService = Executors.newCachedThreadPool();
    
    
        /** * 开启 netty * * @return */
        @RequestMapping("/openNettyServer")
        public CommonResult openNettyServer() { 
    
            try { 
    
                log.info("启动Netty服务，获取可用端口：{}", port);
                Future<Channel> future = executorService.submit(websocketServer);
                Channel channel = future.get();
                if (null == channel) { 
                    throw new RuntimeException("netty server open error channel is null");
                }
                while (!channel.isActive()) { 
                    log.info("启动Netty服务，循环等待启动...");
                    Thread.sleep(500);
                }
                // 放入 缓存
                CacheUtil.serverInfoMap.put(port, new ServerInfo(NetUtil.getHost(), port, new Date()));
                CacheUtil.serverMap.put(port, websocketServer);
    
                log.info("启动Netty服务，完成：{}", channel.localAddress());
                return CommonResult.success();
            } catch (Exception e) { 
                log.error("启动Netty服务失败", e);
                return CommonResult.failed(e.getMessage());
            }
        }
    
        /** * 关闭 netty * * @return */
        @RequestMapping("/closeNettyServer")
        public CommonResult closeNettyServer() { 
            try { 
                log.info("关闭Netty服务开始，端口：{}", port);
                WebsocketServer websocketServer = CacheUtil.serverMap.get(port);
                if (null == websocketServer) { 
                    CacheUtil.serverMap.remove(port);
                    return CommonResult.success();
                }
                websocketServer.destroy();
                CacheUtil.serverMap.remove(port);
                CacheUtil.serverInfoMap.remove(port);
                log.info("关闭Netty服务完成，端口：{}", port);
                return CommonResult.success();
            } catch (Exception e) { 
                log.error("关闭Netty服务失败，端口：{}", port, e);
                return CommonResult.failed();
            }
        }
    
        /** * 查 服务端 列表 * @return */
        @RequestMapping("/queryNettyServerList")
        public Collection<ServerInfo> queryNettyServerList() { 
            try { 
                Collection<ServerInfo> serverInfos = CacheUtil.serverInfoMap.values();
                log.info("查询服务端列表。{}", JSON.toJSONString(serverInfos));
                return serverInfos;
            } catch (Exception e) { 
                log.info("查询服务端列表失败。", e);
                return null;
            }
        }
    
        /** * 从 redis 查 用户管道 * @return */
        @RequestMapping("/queryUserChannelInfoList")
        public List<UserChannelInfo> queryUserChannelInfoList() { 
            try { 
                log.info("查询用户列表信息开始");
                List<UserChannelInfo> userChannelInfoList = redisUtil.popList();
                log.info("查询用户列表信息完成。list：{}", JSON.toJSONString(userChannelInfoList));
                return userChannelInfoList;
            } catch (Exception e) { 
                log.error("查询用户列表信息失败", e);
                return null;
            }
        }
    
    }

**application.yml** 配置文件

server:
      port: 8080
    netty:
      port: 9000
    
    spring:
      application:
        name: nettywebsocket
      redis:
        database: 1
        timeout: 6000   #超时时间
        host: 127.0.0.1
        password: 123456
        port: 6379
        lettuce:        # 这里标明使用lettuce配置
          pool:
            max‐idle: 50              # 连接池中的最大空闲连接
            min‐idle: 10              # 连接池中的最小空闲连接
            max‐active: 100           # 连接池最大连接数（使用负值表示没有限制）
            max‐wait: 1000            #  连接池最大阻塞等待时间（使用负值表示没有限制）

## 4. 测试 ##

可以用index页面，也可以直接 在线测试 [websocket在线测试][websocket]

index.html

<!DOCTYPE HTML>
    <html>
    <head>
        <title>My WebSocket</title>
    </head>
    
    <body>
    <input id="text" type="text"/>
    <button onclick="send()">Send</button>
    <button onclick="closeWebSocket()">Close</button>
    <div id="message"></div>
    </body>
    
    <script type="text/javascript">
        var websocket = null;
    
        //判断当前浏览器是否支持WebSocket, 主要此处要更换为自己的地址
        if ('WebSocket' in window) { 
            websocket = new WebSocket("ws://127.0.0.1:9000/ws");    // netty 端口
        } else { 
            alert('Not support websocket')
        }
    
        //连接发生错误的回调方法
        websocket.onerror = function () { 
            setMessageInnerHTML("error");
        };
    
        //连接成功建立的回调方法
        websocket.onopen = function (event) { 
            //setMessageInnerHTML("open");
        }
    
        //接收到消息的回调方法
        websocket.onmessage = function (event) { 
            console.log("ok i am receive " + event.data);
            setMessageInnerHTML(event.data);
        }
    
        //连接关闭的回调方法
        websocket.onclose = function () { 
            setMessageInnerHTML("close");
        }
    
        //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
        window.onbeforeunload = function () { 
            websocket.close();
        }
    
        //将消息显示在网页上
        function setMessageInnerHTML(innerHTML) { 
            document.getElementById('message').innerHTML += 'receive message is : ' + innerHTML + '<br/>';
        }
    
        //关闭连接
        function closeWebSocket() { 
            websocket.close();
        }
    
        //发送消息
        function send() { 
            var message = document.getElementById('text').value;
            websocket.send(message);
        }
    </script>
    </html>

启动2个实例,，分别调用接口开启netty服务端，连接上线，触发上线事件，便会在redis和缓存中存入信息。然后在线测试

消息体，json格式，根据toChannelId判断是否需要pub
    { "content":"xxxx","toChannelId":"58f5ef52"}

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

http://localhost:8080/index.html  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 2]

## 5.结语 ##

**世上无难事，只  
怕有心人，每天积累一点点，fighting！！！  
2021，加油，fighting，希望可以少些crud啦！**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70]: /images/20221022/a2da40e8f999461aa23e65e5c801930d.png
[websocket]: http://www.websocket-test.com/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 1]: /images/20221022/e3ff3002a1f847468bdb2d37bb45f5ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQzNzYzMw_size_16_color_FFFFFF_t_70 2]: /images/20221022/6f44e1a7ac0041b0bc693e856a5e3a27.png