channelRead对于耗时业务逻辑处理的优化

Posted 2022-09-04 falcon-fei

channelRead对于耗时业务逻辑处理的优化

背景：之前在channelRead中，接收到远端消息进行解码后直接使用了操作数据库这种耗时较久的业务逻辑处理。导致本地netty的工作线程阻塞，会降低可用线程数。另一个对于当前channel的心跳机制也有影响，会导致远端机器长时间接受不到心跳信号，认为这台机器挂掉了。。。

原始代码

       public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
        if(msg instanceof MessageEntity)
        
            MessageEntity messageEntity = (MessageEntity) msg;
            if (2==messageEntity.getMsgType())
            
                LOGGER.info("收到服务端发来的方法请求了--------------------------------------------");
                // 转换为MethodInvokeMeta
                MethodInvokeMeta invokeMeta = (MethodInvokeMeta) ((MessageEntity) msg).getData();
                LOGGER.info(" -> [客户端信息] \n 方法名  - >  \n 参数列表  ->  \n " +
                                "返回值  ->   ", this.getClass().getName(), invokeMeta.getMethodName(), invokeMeta.getArgs()
                        , invokeMeta.getReturnType());
                // 具体的处理类
                RequestDispatcher requestDispatcher = new RequestDispatcher();
                requestDispatcher.dispatcher(ctx, invokeMeta);
            
            else
            
                LOGGER.error("接受到的服务端请求无法识别");
            
        else
        
            LOGGER.error("接受到的服务端请求无法识别");
        
    

在channelRead中，dispatch是一个接受到远程调用请求的分发器，会根据调用参数执行本地具体的方法。其中大多数都包括耗时较久的数据库操作，因此这块代码亟需优化。

    requestDispatcher.dispatcher(ctx, invokeMeta); 

解决方案

啥都不说，先上代码，如下：

    ExecutorService executor = Executors.newFixedThreadPool(2);

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
        //引入异步业务线程池的方式，避免长时间业务耗时业务阻塞netty本身的worker工作线程
        executor.submit(new Callable<Object>() 
            @Override
            public Object call() throws Exception 
                if(msg instanceof MessageEntity)
                
                    MessageEntity messageEntity = (MessageEntity) msg;
                    if (2==messageEntity.getMsgType())
                    
                        LOGGER.info("收到服务端发来的方法请求了--------------------------------------------");
                        // 转换为MethodInvokeMeta
                        MethodInvokeMeta invokeMeta = (MethodInvokeMeta) ((MessageEntity) msg).getData();
                        LOGGER.info(" -> [客户端信息] \n 方法名  - >  \n 参数列表  ->  \n " +
                                        "返回值  ->   ", this.getClass().getName(), invokeMeta.getMethodName(), invokeMeta.getArgs()
                                , invokeMeta.getReturnType());
                        // 具体的处理类
                        RequestDispatcher requestDispatcher = new RequestDispatcher();
                        requestDispatcher.dispatcher(ctx, invokeMeta);
                    
                    else
                    
                        LOGGER.error("接受到的服务端请求无法识别");
                    
                else
                
                    LOGGER.error("接受到的服务端请求无法识别");
                
                return null;
            
        );
    

通过自己添加业务线程池的方式，避免阻塞worker工作线程（因为读完数据后，ChannelPipeline会触发 ChannelHandler链来处理业务逻辑，而ChannelHandler链的整个过程是是同步的）