3

我在让 netty 与 UDP 一起工作时遇到了麻烦。最大的问题是，一旦我与服务器建立连接并完成服务器和客户端之间的交互，服务器就会变得无用。我无法从同一个客户端或任何其他（不同的主机）建立任何其他连接。我觉得他们是我想念的非常简单和容易的东西。我已将服务器配置为使用以下代码为连接到它的每个新主机创建一个新管道（我认为？）：

public class DistinctChannelPipelineFactory implements ChannelPipelineFactory {

  private final ChannelPipelineFactory pipelineFactory;

  public DistinctChannelPipelineFactory(ChannelPipelineFactory pipelineFactory) {
    this.pipelineFactory = pipelineFactory;
  }

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

}

使用 DistinctChannelPipelineHandler 看这个，我尝试为每个远程主机创建一个不同的管道，并在 10 秒后将它们超时。

    private final LoadingCache<SocketAddress, ChannelPipeline> pipelines;

  public DistinctChannelPipelineHandler(ChannelPipelineFactory factory) {
    this.pipelines = CacheBuilder.newBuilder()
        .concurrencyLevel(1)
        .expireAfterAccess(10, SECONDS)
        .removalListener(new PipelineRemovalListener())
        .build(new PipelineCacheLoader(factory));
  }

  public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
    if (e instanceof MessageEvent) {
      final ChannelPipeline pipeline = pipelines.get(((MessageEvent) e).getRemoteAddress());
      if (!pipeline.isAttached()) {
        pipeline.attach(ctx.getChannel(), ctx.getPipeline().getSink());
        pipeline.sendUpstream(new UpstreamChannelStateEvent(ctx.getChannel(), OPEN, TRUE));
      }
      pipeline.sendUpstream(e);
    }

    if (e instanceof ChannelStateEvent) {
      for (final ChannelPipeline pipeline : pipelines.asMap().values()) {
        final ChannelStateEvent cse = (ChannelStateEvent) e;
        pipeline.sendUpstream(new UpstreamChannelStateEvent(ctx.getChannel(), cse.getState(), cse.getValue()));
      }
    }
  }

  public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
    if (e instanceof MessageEvent) {
      final ChannelPipeline pipeline = pipelines.get(((MessageEvent) e).getRemoteAddress());
      if (!pipeline.isAttached()) {
        pipeline.attach(ctx.getChannel(), ctx.getPipeline().getSink());
      }
      pipeline.sendDownstream(e);
    } else {
      ctx.sendDownstream(e);
    }
  }

  private static final class PipelineCacheLoader extends CacheLoader<SocketAddress, ChannelPipeline> {

    private final ChannelPipelineFactory factory;

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

    @Override
    public ChannelPipeline load(SocketAddress key) throws Exception {
      return factory.getPipeline();
    }
  }

  private static final class PipelineRemovalListener implements RemovalListener<SocketAddress, ChannelPipeline> {

    private static final Logger logger = LoggerFactory.getLogger(PipelineRemovalListener.class);

    @Override
    public void onRemoval(RemovalNotification<SocketAddress, ChannelPipeline> n) {
      logger.info("UDP connection timed out, removing connection for {}", n.getKey());
      n.getValue().sendUpstream(new UpstreamChannelStateEvent(n.getValue().getChannel(), OPEN, FALSE));
    }
  }

这就是我初始化服务器的方式：

@Provides
  public ConnectionlessBootstrap getConnectionlessBootstrap(DatagramChannelFactory channelFactory,
                                                            @LocalAddress SocketAddress localAddress,
                                                            final UdpPipelineFactory pipelineFactory) {

    final ConnectionlessBootstrap bootstrap = new ConnectionlessBootstrap(channelFactory);
    bootstrap.setOption("localAddress", localAddress);
    bootstrap.setPipelineFactory(new DistinctChannelPipelineFactory(pipelineFactory));
    return bootstrap;
  }

@Provides
  @Singleton
  public DatagramChannelFactory getDatagramChannelFatory(@WorkerExecutor Executor worker) {
    final DatagramChannelFactory channelFactory = new NioDatagramChannelFactory(worker);
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override public void run() {
        channelFactory.releaseExternalResources();
      }
    });
    return channelFactory;
  }

我省略了我实际添加所有处理程序的位置，因为我认为这不是问题所在。我在这里错过了一些基本的东西吗？我只想要一个超时的每个唯一远程地址的管道。启动服务器并让它仅用于客户端/服务器交互，这非常令人沮丧！我已经通过调试验证了，一旦我用额外的请求点击它，它就不会创建新的管道。因此，原始管道似乎处于非常陈旧的状态，这就是它不接受任何其他请求的原因。想法？建议？

4

1 回答 1

2

犯了一个根本性的错误。使用 ConnectionlessBootstrap 一切都在同一个通道上运行，我们在每次调用服务器后关闭通道......从而禁用​​ UDP。这就是我们的 TCP 代码所做的事情，并且花了一段时间才意识到它的工作方式不同。希望其他人可以节省一些时间和头痛。

于 2013-08-09T19:47:55.870 回答