Netty OrderedMemoryAwareThreadPoolExecutor 没有创建多个线程

Question

我将 Netty 用于多线程 TCP 服务器和单个客户端持久连接。 客户端发送许多二进制消息（在我的用例中为 10000 条），并且应该接收每条消息的答案。我在管道中添加了一个 OrderedMemoryAwareThreadPoolExecutor 来处理在多个线程上执行 DB 调用。

如果我在 messageReceived() 方法中运行 DB 调用（或使用 Thread.currentThread().sleep(50) 模拟它），那么所有事件都由 单线程 处理。 p>

    5 count of {main}
    1 count of {New
10000 count of {pool-3-thread-4}

对于 messageReceived() 的简单实现，服务器会按预期创建许多执行线程。

我应该如何配置 ExecutionHandler 以获得业务逻辑的多个线程执行器？

这是我的代码：

public class MyServer {

      public void run() {
            OrderedMemoryAwareThreadPoolExecutor eventExecutor = new OrderedMemoryAwareThreadPoolExecutor(16, 1048576L, 1048576L, 1000, TimeUnit.MILLISECONDS, Executors.defaultThreadFactory());  
            ExecutionHandler executionHandler = new ExecutionHandler(eventExecutor);        
            bootstrap.setPipelineFactory(new ServerChannelPipelineFactory(executionHandler));
      }
    }  



    public class ServerChannelPipelineFactory implements ChannelPipelineFactory {

      public ChannelPipeline getPipeline() throws Exception {

        pipeline.addLast("encoder", new MyProtocolEncoder());
        pipeline.addLast("decoder", new MyProtocolDecoder());
        pipeline.addLast("executor", executionHandler);
        pipeline.addLast("myHandler", new MyServerHandler(dataSource));

      }
    }

    public class MyServerHandler extends SimpleChannelHandler {

      public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e) throws DBException {


          // long running DB call simulation
          try {
            Thread.currentThread().sleep(50);
          } catch (InterruptedException ex) {

          }  

          // a simple message  
          final MyMessage answerMsg = new MyMessage();
          if (e.getChannel().isWritable()) {
            e.getChannel().write(answerMsg);
          }  
      }      
    }

Answer 1

OrderedMemoryAwareThreadPoolExecutor 保证来自单个通道的事件按顺序处理。您可以将其视为将通道绑定到池中的特定线程，然后处理该线程上的所有事件 - 尽管它比这更复杂一些，因此不要依赖始终由同一线程处理的通道。

如果您启动第二个客户端，您会看到它（很可能）正在池中的另一个线程上进行处理。如果您真的可以并行处理单个客户端的请求，那么您可能需要 MemoryAwareThreadPoolExecutor，但请注意，这不能保证通道事件的顺序。