Netty根据UDP数据报的不同流水线

Question

我们已经有了一个已经在TCP/IP中实现的服务器，但是现在我们也需要支持UDP的协议。

发送的每个UDP数据报都包含我需要解码的所有内容，因此它是一个非常简单的答复和响应系统，数据报中的数据由换行符分隔。

用于当所述服务器启动如下所示的自举代码：

//SETUP UDP SERVER 
    DatagramChannelFactory udpFactory = new NioDatagramChannelFactory(Executors.newCachedThreadPool()); 

    ConnectionlessBootstrap udpBootstrap = new ConnectionlessBootstrap(udpFactory); 

    udpBootstrap.setOption("sendBufferSize", 65536); 
    udpBootstrap.setOption("receiveBufferSize", 65536); 
    udpBootstrap.setOption("receiveBufferSizePredictorFactory", new AdaptiveReceiveBufferSizePredictorFactory()); 

    udpBootstrap.setOption("broadcast", "true"); 
    udpBootstrap.setPipelineFactory(new ServerPipeLineFactoryUDP()); 
    udpBootstrap.bind(new InetSocketAddress(hostIp, 4000)); 

流水线的代码是：

class ServerPipeLineFactoryUDP implements ChannelPipelineFactory 
{ 

    private final static ExecutionHandler EXECUTION_HANDLER = new ExecutionHandler(new OrderedMemoryAwareThreadPoolExecutor(ScorpionFMS.THREAD_POOL_COUNT, 0, 0)); 

    public ServerPipeLineFactoryUDP() 
    { 

    } 

    @Override 
    public ChannelPipeline getPipeline() throws Exception 
    { 

    ChannelPipeline pipeline = pipeline(); 
    pipeline.addLast("debugup", new DebugUpstreamHandler("UDP")); 
    pipeline.addLast("debugdown", new DebugDownstreamHandler("UDP")); 

    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(256, Delimiters.lineDelimiter())); 

    pipeline.addLast("decoder", new UDPRequestDecoder(true)); 
    pipeline.addLast("encoder", new StringEncoder()); 
    pipeline.addLast("executor", EXECUTION_HANDLER); 
    pipeline.addLast("handler", new UDPRequestHandler(

    return pipeline; 
    } 
} 

问题IM具有是每个数据报被使用的相同的实例这个流水线（我希望每个数据报将使用一个新的流水线实例），所以我在处理数据报内容时所存储的所有状态都会被保存下来，并且下一个数据报也会使用它（而对于TCP，每个连接都会有它的自己的渠道，因此它自己的我管道的状态和它自己的状态）

我知道这是从阅读文档中的预期行为，但无论如何强迫网络重新创建每个数据报的管道？或者我是否完全错误的方式？

说得简洁，我想每个数据报有管道（同TCP）

Answer 1

就像我在IRC中所说的，我认为这可以做你想做的或者至少给你一些想法。

public class Example { 

    public static void main(String[] args) { 
     final ChannelPipelineHandlerImpl perDatagramFactory = new ChannelPipelineHandlerImpl(); 

     DatagramChannelFactory udpFactory = new NioDatagramChannelFactory(Executors.newCachedThreadPool()); 

     ConnectionlessBootstrap udpBootstrap = new ConnectionlessBootstrap(udpFactory); 

     udpBootstrap.setPipelineFactory(new ChannelPipelineFactory() { 

      public ChannelPipeline getPipeline() throws Exception { 
       return Channels.pipeline(new DistinctChannelPipelineHandler(perDatagramFactory)); 
      } 
     }); 

    } 

    private static final class DistinctChannelPipelineHandler implements ChannelDownstreamHandler, ChannelUpstreamHandler { 
     private ChannelPipelineFactory factory; 

     public DistinctChannelPipelineHandler(ChannelPipelineFactory factory) { 
      this.factory = factory; 
     } 

     public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception { 
      ChannelPipeline pipeline = factory.getPipeline(); 
      pipeline.attach(ctx.getChannel(), ctx.getPipeline().getSink()); 
      pipeline.sendUpstream(e); 

      ctx.sendUpstream(e); 

     } 

     public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception { 
      ChannelPipeline pipeline = factory.getPipeline(); 
      pipeline.attach(ctx.getChannel(), ctx.getPipeline().getSink()); 
      pipeline.sendDownstream(e); 

      ctx.sendDownstream(e); 
     } 

    } 

    private static final class ChannelPipelineHandlerImpl implements ChannelPipelineFactory { 

     public ChannelPipeline getPipeline() throws Exception { 
      // Add your handlers here 
      return Channels.pipeline(); 
     } 

    } 
} 

Answer 2

的新实例，我不知道对UDP通道是如何被处理，但如果渠道是不同的每个数据报，您可以将您的状态存储在ChannelLocal s中。