multithreading - 使用 Netty 在 UDP 服务器中丢失了大量 UDP 请求

Question

我用 Netty 编写了一个简单的 UDP 服务器，它只是在日志中打印出接收到的消息(帧)。为此，我创建了一个简单的帧解码器解码器和一个简单的消息处理程序。我还有一个客户端可以顺序和/或并行发送多个请求。

当我配置我的客户端测试器以顺序发送几百个请求时，它们之间有一个小的延迟，我用 Netty 编写的服务器正确接收它们。但是目前我增加了客户端中同时请求的数量(例如 100 个)加上顺序请求和重复请求的数量，我的服务器开始丢失许多请求。例如，当我发送 50000 个请求时，仅使用简单的 ChannelHandler 打印出接收到的消息时，我的服务器只收到大约 49000 个请求。

当我在这个处理程序前面添加简单的帧解码器(打印出帧并将其复制到另一个缓冲区)时，服务器只处理一半的请求!!

我注意到无论我为创建的 NioDatagramChannelFactory 指定的工作线程数量如何，始终只有一个线程处理请求(我使用推荐的 Executors.newCachedThreadPool() 作为另一个参数)。

我还基于 JDK 附带的 DatagramSocket 创建了另一个类似的简单 UDP 服务器，它可以完美地处理每个请求，而 0(零)丢失!当我在我的客户端发送 50000 个请求(例如有 1000 个线程)时，我在我的服务器中收到了 50000 个请求。

使用 Netty 配置我的 UDP 服务器时我做错了什么吗？或者，也许 Netty 根本就不是为支持这种负载而设计的？？为什么给定的缓存线程池只使用了一个线程(我注意到通过查看 JMX jconsole 和检查输出日志中的线程名称，只有一个线程使用并且始终相同)？我认为如果按预期使用更多线程，服务器将能够轻松处理此类负载，因为在不使用 Netty 时我可以毫无问题地做到这一点!

看下面我的初始化代码:

...

lChannelfactory = new NioDatagramChannelFactory( Executors.newCachedThreadPool(), nbrWorkers );
lBootstrap = new ConnectionlessBootstrap( lChannelfactory );

lBootstrap.setPipelineFactory( new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline()
    {
        ChannelPipeline lChannelPipeline = Channels.pipeline();
        lChannelPipeline.addLast( "Simple UDP Frame Dump DECODER", new SimpleUDPPacketDumpDecoder( null ) );            
        lChannelPipeline.addLast( "Simple UDP Frame Dump HANDLER", new SimpleUDPPacketDumpChannelHandler( lOuterFrameStatsCollector ) );            
        return lChannelPipeline;
    }
} );

bindChannel = lBootstrap.bind( socketAddress );

...

以及我的解码器中 decode() 方法的内容:

protected Object decode(ChannelHandlerContext iCtx, Channel iChannel, ChannelBuffer iBuffer) throws Exception
{
    ChannelBuffer lDuplicatedChannelBuffer = null;
    sLogger.debug( "Decode method called." );

    if ( iBuffer.readableBytes() < 8 ) return null;
    if ( outerFrameStatsCollector != null ) outerFrameStatsCollector.incrementNbrRequests();

    if ( iBuffer.readable() ) 
    {        
        sLogger.debug( convertToAsciiHex( iBuffer.array(), iBuffer.readableBytes() ) );                     
        lDuplicatedChannelBuffer = ChannelBuffers.dynamicBuffer( iBuffer.readableBytes() );            
        iBuffer.readBytes( lDuplicatedChannelBuffer );
    }

    return lDuplicatedChannelBuffer;
}

我的处理程序中 messageReceived() 方法的内容:

public void messageReceived(final ChannelHandlerContext iChannelHandlerContext, final MessageEvent iMessageEvent) throws Exception
{
    ChannelBuffer lMessageBuffer = (ChannelBuffer) iMessageEvent.getMessage();
    if ( outerFrameStatsCollector != null ) outerFrameStatsCollector.incrementNbrRequests();

    if ( lMessageBuffer.readable() ) 
    {        
        sLogger.debug( convertToAsciiHex( lMessageBuffer.array(), lMessageBuffer.readableBytes() ) );            
        lMessageBuffer.discardReadBytes();
    }
}

Answer 1

您没有正确配置 ConnectionlessBootstrap 实例。

您必须使用最佳值配置以下内容。

SO_SNDBUF 大小、SO_RCVBUF 大小和 ReceiveBufferSizePredictorFactory

lBootstrap.setOption("sendBufferSize", 1048576);

lBootstrap.setOption("receiveBufferSize", 1048576);

lBootstrap.setOption("receiveBufferSizePredictorFactory", 
 new AdaptiveReceiveBufferSizePredictorFactory(MIN_SIZE, INITIAL_SIZE, MAX_SIZE));

检查 DefaultNioDatagramChannelConfig 类以获取更多详细信息。

管道使用 Netty 工作线程完成所有工作。如果
工作线程过载，会延迟选择器事件循环
执行，读/写会出现瓶颈
channel 。您必须添加一个执行处理程序，如下所示
管道。它将释放工作线程来做它自己的工作。

ChannelPipeline lChannelPipeline = Channels.pipeline();

lChannelPipeline.addFirst("execution-handler", new ExecutionHandler(
  new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576));

//add rest of the handlers here