Netty比Tomcat慢

Question

我们刚刚完成构建服务器以将数据存储到磁盘并使用Netty进行了前端处理.在负载测试期间,我们看到Netty扩展到每秒约8,000条消息.鉴于我们的系统,这看起来很低.对于基准测试,我们编写了一个Tomcat前端并运行相同的负载测试.通过这些测试,我们每秒获得大约25,000条消息.

以下是我们的负载测试机的规格:

• Macbook Pro Quad核心
• 16GB的RAM
• Java 1.6

这是Netty的负载测试设置:

• 10个主题
• 每个线程100,000条消息
• Netty服务器代码(非常标准) - 服务器上的Netty管道是两个处理程序:FrameDecoder和处理请求和响应的SimpleChannelHandler.
• 客户端JIO使用Commons Pool来池和重用连接(池的大小与线程数相同)

以下是Tomcat的负载测试设置:

• 10个主题
• 每个线程100,000条消息
• Tomcat 7.0.16,默认配置使用Servlet来调用服务器代码
• 客户端使用URLConnection而没有任何池

我的主要问题是为什么表现如此巨大？有没有明显的Netty可以让它比Tomcat运行得更快？

编辑:这是主要的Netty服务器代码:

NioServerSocketChannelFactory factory = new NioServerSocketChannelFactory();
ServerBootstrap server = new ServerBootstrap(factory);
server.setPipelineFactory(new ChannelPipelineFactory() {
  public ChannelPipeline getPipeline() {
    RequestDecoder decoder = injector.getInstance(RequestDecoder.class);
    ContentStoreChannelHandler handler = injector.getInstance(ContentStoreChannelHandler.class);
    return Channels.pipeline(decoder, handler);
  }
});

server.setOption("child.tcpNoDelay", true);
server.setOption("child.keepAlive", true);
Channel channel = server.bind(new InetSocketAddress(port));
allChannels.add(channel);

我们的处理程序看起来像这样:

public class RequestDecoder extends FrameDecoder {
  @Override
  protected ChannelBuffer decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) {
    if (buffer.readableBytes() < 4) {
      return null;
    }

    buffer.markReaderIndex();
    int length = buffer.readInt();
    if (buffer.readableBytes() < length) {
      buffer.resetReaderIndex();
      return null;
    }

    return buffer;
  }
}

public class ContentStoreChannelHandler extends SimpleChannelHandler {
  private final RequestHandler handler;

  @Inject
  public ContentStoreChannelHandler(RequestHandler handler) {
    this.handler = handler;
  }

  @Override
  public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
    ChannelBuffer in = (ChannelBuffer) e.getMessage();
    in.readerIndex(4);

    ChannelBuffer out = ChannelBuffers.dynamicBuffer(512);
    out.writerIndex(8); // Skip the length and status code

    boolean success = handler.handle(new ChannelBufferInputStream(in), new ChannelBufferOutputStream(out), new NettyErrorStream(out));
    if (success) {
      out.setInt(0, out.writerIndex() - 8); // length
      out.setInt(4, 0); // Status
    }

    Channels.write(e.getChannel(), out, e.getRemoteAddress());
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
    Throwable throwable = e.getCause();
    ChannelBuffer out = ChannelBuffers.dynamicBuffer(8);
    out.writeInt(0); // Length
    out.writeInt(Errors.generalException.getCode()); // status

    Channels.write(ctx, e.getFuture(), out);
  }

  @Override
  public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) {
    NettyContentStoreServer.allChannels.add(e.getChannel());
  }
}

更新:

我设法让我的Netty解决方案在4,000 /秒以内.几个星期前,我在我的连接池中测试客户端PING作为防止空闲套接字的安全防范,但我忘记在开始加载测试之前删除该代码.每次从池中检出Socket时,此代码都会有效地对服务器进行PING(使用Commons Pool).我评论说代码输出,我现在使用Netty获得21,000 /秒,使用Tomcat获得25,000 /秒.

尽管如此,这对Netty来说是个好消息,但我仍然使用Netty比Tomcat少了4,000 /秒.我可以发布我的客户端(我认为我已经排除但显然不是)如果有人有兴趣看到它.

Answer 1

该方法messageReceived是使用工作线程执行的，该线程可能因RequestHandler#handle忙于执行某些 I/O 工作而被阻塞。您可以尝试在通道管道中添加OrderdMemoryAwareThreadPoolExecutor（推荐）用于执行处理程序，或者尝试将处理程序工作分派到新的ThreadPoolExecutor并传递对套接字通道的引用，以便稍后将响应写回客户端。前任。：

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {   

    executor.submit(new Runnable() {
        processHandlerAndRespond(e);        
    });
}

private void processHandlerAndRespond(MessageEvent e) {

    ChannelBuffer in = (ChannelBuffer) e.getMessage();
    in.readerIndex(4);
    ChannelBuffer out = ChannelBuffers.dynamicBuffer(512);
    out.writerIndex(8); // Skip the length and status code
    boolean success = handler.handle(new ChannelBufferInputStream(in), new ChannelBufferOutputStream(out), new NettyErrorStream(out));
    if (success) {
        out.setInt(0, out.writerIndex() - 8); // length
        out.setInt(4, 0); // Status
    }
    Channels.write(e.getChannel(), out, e.getRemoteAddress());
}