java - 每个远程主机具有不同管道的 UDP 无法正常工作

Question

我在使用 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;
  }

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

Answer 1

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