很多的UDP请求丢失在UDP服务器与Netty

Question

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

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

当我在该处理程序前添加简单帧解码器（打印出帧并将其复制到另一个缓冲区中）时，服务器仅处理一半的请求！我注意到，无论我指定给创建的NioDatagramChannelFactory的工作器的数量是多少，总是只有一个线程处理请求（我使用推荐的Executors.newCachedThreadPool（）作为另一个参数）。

我还创建了另一个基于DatagramSocket与JDK一起的类似的简单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实例。

1. 您必须配置以下最佳值。

   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类的更多细节。

2. 管道正在使用Netty工作线程执行所有操作。如果 工作线程过载，它将延迟执行选择器事件循环 ，并且在读取/写入通道时会出现瓶颈。您必须添加一个执行处理程序，如下面的 管道中所示。它将释放工作线程来完成自己的工作。

   ChannelPipeline lChannelPipeline = Channels.pipeline(); 
   
   lChannelPipeline.addFirst("execution-handler", new ExecutionHandler(
       new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576)); 
   
   //add rest of the handlers here