Netty 实现一个端口同时接收 socket 和 webSocket 连接

亦凉 2022-11-20 12:55 607阅读 0赞

项目是在 Springboot 集成 Netty 的基础上开发。

参考文章：  
1、SpringBoot 整合 Netty 实现Socket：https://www.cnblogs.com/guoyuchuan/p/9581283.html  
2、Netty 实现 webSocket：https://www.cnblogs.com/miller-zou/p/7002070.html  
3、同时实现 Socket && WebScoket：https://cloud.tencent.com/developer/article/1366184

第一部分：DeviceServer

@Component
    public class DeviceServer { 
     
        private static Logger logger = LoggerFactory.getLogger(DeviceServer.class);
       
        @Resource
        private ChannelActiveHandler channelActiveHandler;
        @Resource
        private DeviceServerHandler deviceServerHandler;
     
        //配置服务端线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        ChannelFuture socketfuture = null;
        
        @PreDestroy             //关闭spring容器后释放资源
        public void stop(){ 
            if(socketfuture!=null){ 
                socketfuture.channel().close().addListener(ChannelFutureListener.CLOSE);
                socketfuture.awaitUninterruptibly();
                socketfuture=null;
                logger.info("Netty 服务端关闭");
            }
           
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
     
        /** * 启动流程 */
        public void run(int port) throws InterruptedException { 
     
            try{ 
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workGroup)
                        .channel(NioServerSocketChannel.class)
                        .childOption(ChannelOption.SO_REUSEADDR, true) //快速复用端口
                        .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS,1000)
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception { 
     
                                ch.pipeline().addLast("active",channelActiveHandler);
                                //Socket 连接心跳检测
    			    ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 0, 0));
     
                                ch.pipeline().addLast("socketChoose",new SocketChooseHandler());
                                
    			    //注意，这个专门针对 Socket 信息的解码器只能放在 SocketChooseHandler 之后，否则会导致 webSocket 连接出错
                                ch.pipeline().addLast("myDecoder",new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN,1024*1024,0,4,0,4,true));
                                ch.pipeline().addLast("commonhandler",deviceServerHandler);
                            }
                        });
     
                //绑定端口，同步等待成功
                socketfuture = serverBootstrap.bind(port).sync();
                if(socketfuture.isSuccess()){ 
                    logger.info("Netty 服务已启动");
                }
               
                socketfuture.channel().closeFuture().sync();    
     
            }catch (Exception e){ 
                //优雅退出，释放线程池
                workGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }finally { 
                //优雅退出，释放线程池
                workGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
     
        }
    }

第二部分：ChannelActiveHandler

@Component
    @ChannelHandler.Sharable
    public class ChannelActiveHandler extends ChannelInboundHandlerAdapter { 
        private static Logger logger = LoggerFactory.getLogger(ChannelActiveHandler.class);
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
            String clientIP = insocket.getAddress().getHostAddress();
            String clientPort = String.valueOf(insocket.getPort());
            logger.info("新的连接："+ clientIP +":"+ clientPort);
        }
    }

第三部分：SocketChooseHandler

/** * 协议初始化解码器. * * 用来判定实际使用什么协议.</b> * */
    @Component
    public class SocketChooseHandler extends ByteToMessageDecoder { 
        /** 默认暗号长度为23 */
        private static final int MAX_LENGTH = 23;
        /** WebSocket握手的协议前缀 */
        private static final String WEBSOCKET_PREFIX = "GET /";
        @Resource
        private SpringContextUtil springContextUtil;
     
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
            String protocol = getBufStart(in);
            if (protocol.startsWith(WEBSOCKET_PREFIX)) { 
                springContextUtil.getBean(PipelineAdd.class).websocketAdd(ctx);
     
                //对于 webSocket ，不设置超时断开
                ctx.pipeline().remove(IdleStateHandler.class);
                ctx.pipeline().remove(LengthFieldBasedFrameDecoder.class);
            }
            in.resetReaderIndex();
            ctx.pipeline().remove(this.getClass());
        }
     
        private String getBufStart(ByteBuf in){ 
            int length = in.readableBytes();
            if (length > MAX_LENGTH) { 
                length = MAX_LENGTH;
            }
     
            // 标记读位置
            in.markReaderIndex();
            byte[] content = new byte[length];
            in.readBytes(content);
            return new String(content);
        }
    }

第四部分：SpringContextUtil

@Component
    public class SpringContextUtil implements ApplicationContextAware { 
     
        private static ApplicationContext applicationContext;
     
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 
            SpringContextUtil.applicationContext = applicationContext;
        }
     
        /** * @Description: 获取spring容器中的bean, 通过bean类型获取 */
        public static <T> T getBean(Class<T> beanClass) { 
            return applicationContext.getBean(beanClass);
        }
       
    }

第五部分：PipelineAdd

@Component
    public class PipelineAdd { 
     
        public  void websocketAdd(ChannelHandlerContext ctx){ 
            System.out.println("PipelineAdd");
            // HttpServerCodec：将请求和应答消息解码为HTTP消息
            ctx.pipeline().addBefore("commonhandler","http-codec",new HttpServerCodec());
     
            // HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息
            ctx.pipeline().addBefore("commonhandler","aggregator",new HttpObjectAggregator(65535));
     
            // ChunkedWriteHandler：向客户端发送HTML5文件,文件过大会将内存撑爆
            ctx.pipeline().addBefore("commonhandler","http-chunked",new ChunkedWriteHandler());
     
            ctx.pipeline().addBefore("commonhandler","WebSocketAggregator",new WebSocketFrameAggregator(65535));
     
            //用于处理websocket, /ws为访问websocket时的uri
            ctx.pipeline().addBefore("commonhandler","ProtocolHandler", new WebSocketServerProtocolHandler("/ws"));
     
        }
    }

第六部分：DeviceServerHandler

@Component
    @ChannelHandler.Sharable
    public class DeviceServerHandler extends SimpleChannelInboundHandler<Object> { 
     
        private static Logger logger = LoggerFactory.getLogger(DeviceServerHandler.class);
       
        //由于继承了SimpleChannelInboundHandler，这个方法必须实现，否则报错
        //但实际应用中，这个方法没被调用
        @Override
        public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { 
            ByteBuf buff = (ByteBuf) msg;
            String info = buff.toString(CharsetUtil.UTF_8);
            logger.info("收到消息内容："+info);
        }
     
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            // WebSocket消息处理
            if (msg instanceof WebSocketFrame) { 
                logger.info("WebSocket消息处理************************************************************");
                String webSocketInfo = ((TextWebSocketFrame) msg).text().trim();
                logger.info("收到webSocket消息：" + webSocketInfo);
            }
            // Socket消息处理
            else{ 
                logger.info("Socket消息处理=================================");
                ByteBuf buff = (ByteBuf) msg;
                String socketInfo = buff.toString(CharsetUtil.UTF_8).trim();;
                logger.info("收到socket消息："+socketInfo);
            }
        }
     
        /*******************************************************************************************/
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
            if (evt instanceof IdleStateEvent) { 
                IdleState state = ((IdleStateEvent) evt).state();
                if (state == IdleState.READER_IDLE) { 
                    // 在规定时间内没有收到客户端的上行数据, 主动断开连接
                    socketChannelMap.remove((SocketChannel)ctx.channel());
                    ctx.disconnect();
                    logger.info("心跳检测触发，socket连接断开！");
                }
            } else { 
                super.userEventTriggered(ctx, evt);
            }
        }
     
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
            InetSocketAddress reAddr = (InetSocketAddress) ctx.channel().remoteAddress();
            String clientIP = reAddr.getAddress().getHostAddress();
            String clientPort = String.valueOf(reAddr.getPort());
            logger.info("连接断开："+ clientIP +":"+ clientPort);
        }
     
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
     
    }

首先我们得先弄清楚websocket和tcpSocket的区别。websocket也是基于tcp的应用层协议，只是在传统的socket上进行了封装。  
然后我们要知道netty的handle支持动态增删。  
说明：

首先我们先添加好SocketChooseHandle()，这是我们的handle判断处理器。如果判断协议是以GET /开头的话，那么必定是websocket的连接握手。

而又因为socket连接是不进SocketChooseHandle的破方法的，导致我们必须在初始化的时候就把socket的处理写在后面。

继续说websocket的处理。

当我们检测到时websocket连接的时候，我们会移除掉socket的编解码处理器，然后再移除自己。（下次进来就直接处理websocketframe了，所以不需要再次进行这个判断处理器）

然后websocket顺利的进入动态添加的编码器，进行websocket的握手handshake。然后进行下一轮通信。

反之，如果是tcpsocket连接，会直接走tcpSocketHandle处理。这里我们用的LengthFieldBasedFrameDecoder长度占包粘包处理器。

protocolResolveHandle是我们的业务处理器handle。