26

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

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

  • Macbook Pro 四核
  • 16GB 内存
  • 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（使用 Commons Pool）时，此代码都会有效地 PING 服务器。我注释掉了这段代码，现在我用 Netty 获得了 21,000/秒，使用 Tomcat 获得了 25,000/秒。

尽管这在 Netty 方面是个好消息，但 Netty 的速度仍然比 Tomcat 少 4,000/秒。如果有人有兴趣看到，我可以发布我的客户端（我认为我已经排除但显然没有）。

4

1 回答 1

2

该方法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());
}
于 2016-09-20T14:04:11.877 回答