42、Netty 源码解析 - 任务加入异步线程池

一、基本说明

1、在 Netty 中做耗时的,不可预料的操作,比如数据库、网络请求,会严重影响 Netty 对 Socket 的处理速度。
2、解决方案就是将耗时任务添加到异步线程池中。但是就添加线程池这步操作来讲,可以有2种方式:

  • 处理耗时业务的第一种方式----handler中加入线程池
  • 处理耗时业务的第二种方式----Context中添加线程池

二、handler中加入线程池

2.1 业务代码

@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    // group 充当业务线程池,可以将任务提交到该线程池中
    // 这里创建了 16 个线程
    static final EventExecutorGroup group = new DefaultEventExecutorGroup(16);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {

        System.out.println("EchoServerHandler 的线程是=" + Thread.currentThread().getName());

        // 将任务提交到 group 线程池
        group.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                // 接收客户端消息
                ByteBuf buf = (ByteBuf) msg;
                byte[] bytes = new byte[buf.readableBytes()];
                buf.readBytes(bytes);
                String body = new String(bytes, "UTF-8");
                // 休眠 10 秒
                Thread.sleep(10 * 1000);
                System.out.println("group.submit 的 call 的线程是=" + Thread.currentThread().getName());
                // 回复消息
                ctx.writeAndFlush(Unpooled.copiedBuffer("Hello,客户端222", CharsetUtil.UTF_8));
                return null;
            }
        });
        
        System.out.println("go on");
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

在channelRead 方法中,模拟了一个耗时 10秒的操作,这里我们将这个任务提交到了一个自定义的业务线程池(group)中,这样,就不会阻塞 Netty 的 IO 线程。

2.2 整个程序处理过程

 

1、当 IO 线程轮询到一个 socket 事件后,交给 IO 线程开始处理,当走到耗时 handler 的时候,将耗时任务交给业务线程池。
2、当耗时任务执行完毕再执行 pipeline write 方法的时候(代码中使用的是ctx.writeAndFlush),会将这个任务交给 IO 线程

2.3 write方法源码(AbstractChannelHandlerContext)

private void write(Object msg, boolean flush, ChannelPromise promise) {
    AbstractChannelHandlerContext next = findContextOutbound();
    final Object m = pipeline.touch(msg, next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        if (flush) {
            next.invokeWriteAndFlush(m, promise);
        } else {
            next.invokeWrite(m, promise);
        }
    } else {
        AbstractWriteTask task;
        if (flush) {
            task = WriteAndFlushTask.newInstance(next, m, promise);
        }  else {
            task = WriteTask.newInstance(next, m, promise);
        }
        safeExecute(executor, task, promise, m);
    }
}

1、当判断下个 outbound 的 executor 线程不是当前线程的时候,会将当前的工作封装成 task,然后放入队列中,等待 IO 任务执行完毕后执行队列中的任务。
2、当我们使用了group.submit(new Callable() {},在 handler 中加入线程池,就会进入到 safeExecute(executor, task, promise, m); 如果去掉这段代码,而使用普通方式来执行耗时的业务,那么就不会进入到 safeExecute(executor, task, promise, m);

三、Context中加入线程池

3.1 业务代码

public final class EchoServer {

    static final boolean SSL = System.getProperty("ssl") != null;
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    // 创建业务线程池
    // 这里创建了有两个线程的线程池
    static final EventExecutorGroup group = new DefaultEventExecutorGroup(2);

    public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     //p.addLast(new EchoServerHandler());
                     // 说明:如果在 addLast 添加 handler 时,前面有指定
                     // EventExecutorGroup ,那么该 handler 会优先加入到该线程池中
                     p.addLast(group,new EchoServerHandler());
                 }
             });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

1、handler 中的代码就使用普通的方式来处理耗时业务。
2、当我们在调用 addLast 方法添加线程池后,handler 将优先使用这个线程池,如果不添加,将使用 IO 线程。
3、当走到 AbstractChannelHandlerContext 的 invokeChannelRead 方法的时候,executor.inEventLoop() 是不会通过的,因为当前线程是 IO 线程,Context(也就是Handler)的 executor 是业务线程,所以会异步执行。

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
     final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
      EventExecutor executor = next.executor();
      if (executor.inEventLoop()) {
          next.invokeChannelRead(m);
      } else {
          executor.execute(new Runnable() {
              @Override
              public void run() {
                  next.invokeChannelRead(m);
              }
          });
      }
  }

4、验证时,如果去掉 p.addLast(group,new EchoServerHandler()); 改为 p.addLast(new EchoServerHandler()); 就不会进行异步执行

四、两种方式的比较

1、第一种方式在 handler 中添加异步,可能更加的自由,比如如果需要访问数据库,那就异步,如果不需要,那就不异步。异步会拖长接口响应时间,因为需要将任务放进 mpscTask 中。如果 IO 时间很短,task 很多,可能一个循环下来,都没时间执行整个 task,导致响应时间达不到指标
2、第二种方式是 Netty 标准方式(即加入到队列),但是,这么做会将整个 handler 都交给业务线程池。不论耗时不耗时,都加入到队列,不够灵活