如何使用netty在单独的线程池中执行业务逻辑处理程序

我有一个需要执行一些业务逻辑的处理程序，我希望它在单独的线程池中执行，以免阻塞 io 事件循环。我已将 DefaultEventExecutorGroup 添加到管道中，如中指定的http://netty.io/4.0/api/io/netty/channel/ChannelPipeline.htmljavadoc 和http://netty.io/wiki/new-and-noteworthy-in-4.0.html#no-more-executionhandler---its-in-the-core wiki:

ch.pipeline().addLast(new DefaultEventExecutorGroup(10), new ServerHandler());

出于测试目的，我的 ServerHandler 只是让当前线程休眠 5 秒：

protected void channelRead0(ChannelHandlerContext ctx, Command cmd) throws Exception {
    System.out.println("Starting.");

    try {
        Thread.currentThread().sleep(5000);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    System.out.println("Finished.");        
}

但显然业务逻辑仍然是同步执行的：

Starting.
Finished.
Starting.
Finished.
Starting.
Finished.

我缺少什么？

如果您的目标不是阻止 IO 事件循环 - 您做对了。但由于 netty 特定，您的处理程序将始终附加到 EventExecutorGroup 的同一线程，因此您上面描述的行为是预期的。

如果您想在阻塞操作到达时立即并行执行，则需要使用另一种方式 - 单独ThreadPoolExecutor。像这样：

ch.pipeline().addLast(new ServerHandler(blockingThreadPool));

where blockingThreadPool是有规律的ThreadPoolExecutor.

例如：

ExecutorService blockingThreadPool = Executors.newFixedThreadPool(10);

现在，在逻辑处理程序中，您可以向此执行程序提交阻塞任务，如下所示：

protected void channelRead0(ChannelHandlerContext ctx, Command cmd) throws Exception {

    blockingIOProcessor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.println("Starting.");

            try {
                Thread.currentThread().sleep(5000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }

            System.out.println("Finished.");
        }
    });

}

您还可以将上下文传递给此可运行对象，以便在处理完成时返回响应（如果需要）。