Netty + NioDatagram(UDP) + ProtoBuf 问题 - 丢弃入站消息 DatagramPacket

Question

我正在尝试使用带有 NioDatagramChannel 和 protoBuf 作为消息的 netty 构建一个 UDP 服务器。这是服务器实现：

EventLoopGroup group = new NioEventLoopGroup();
try {
  Bootstrap b = new Bootstrap();
  b.group(group)
      .channel(NioDatagramChannel.class)
      .option(ChannelOption.SO_BROADCAST, true)
      .handler(new UdpProtoBufServerChannelInitializer());

  b.bind(port).sync().channel().closeFuture().await();
} finally {
  group.shutdownGracefully();
}

UdpProtoBufServerChannelInitializer:

public class UdpProtoBufServerChannelInitializer extends ChannelInitializer<NioDatagramChannel> {

    static final Logger logger = Logger.getLogger(UdpProtoBufServerChannelInitializer.class);
    @Override
      protected void initChannel(NioDatagramChannel ch) throws Exception {
        ChannelPipeline p = ch.pipeline();
        p.addFirst(new ProtobufVarint32FrameDecoder());
        p.addFirst(new ProtobufDecoder(Message.getDefaultInstance()));
        p.addFirst(new ProtobufVarint32LengthFieldPrepender());
        p.addFirst(new ProtobufEncoder());

        p.addFirst(new UdpProtoBufServerHandler());
    }
  }

但是，在运行 UdpProtoBufServerChannelInitializer 之前收到来自 java 客户端的消息后，我看到以下错误。

DEBUG DefaultChannelPipeline:76 - 丢弃入站消息 DatagramPacket(/127.0.0.1:60556 => 0.0.0.0/0.0.0.0:55559, PooledUnsafeHeapByteBuf(ridx: 0, widx: 34, cap: 2048))到达管道的尾部。请检查您的管道配置。

另外，不确定启动应用程序时的这个异常是否相关：

    2017-05-22 14:31:09 DEBUG PlatformDependent0:91 - jdk.internal.misc.Unsafe.allocateUninitializedArray(int): unavailable
java.lang.ClassNotFoundException: jdk.internal.misc.Unsafe
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at io.netty.util.internal.PlatformDependent0.run(PlatformDependent0.java:295)
    at java.security.AccessController.doPrivileged(Native Method)
    at io.netty.util.internal.PlatformDependent0.<clinit>(PlatformDependent0.java:288)
    at io.netty.util.internal.PlatformDependent.getSystemClassLoader(PlatformDependent.java:895)
    at io.netty.util.internal.PlatformDependent.isAndroid0(PlatformDependent.java:919)
    at io.netty.util.internal.PlatformDependent.<clinit>(PlatformDependent.java:70)
    at io.netty.util.ConstantPool.<init>(ConstantPool.java:32)
    at io.netty.util.Signal.<init>(Signal.java:27)
    at io.netty.util.Signal.<clinit>(Signal.java:27)
    at io.netty.util.concurrent.DefaultPromise.<clinit>(DefaultPromise.java:43)
    at io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:36)
    at io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:58)
    at io.netty.channel.MultithreadEventLoopGroup.<init>(MultithreadEventLoopGroup.java:52)
    at io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:87)
    at io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:82)
    at io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:63)
    at io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:51)
    at io.netty.channel.nio.NioEventLoopGroup.<init>(NioEventLoopGroup.java:43)

Answer 1

嗯，看起来你的管道没有消费者。您应该将自定义处理程序放在处理数据的位置。您可以在 p.addFirst(new ProtobufVarint32FrameDecoder()); 之前使用 p.addFirst(new MyHandler()); 或在 p.addFirst(new UdpProtoBufServerHandler()); 之后使用 p.addLast(new MyHandler());.

事实上，我可以假设您想在代码中使用 addLast 而不是 addFirst。如果您使用 addLast，您将获得在管道 ProtobufVarint32FrameDecoder -> ProtobufDecoder -> ... -> UdpProtoBufServerHandler 中声明的顺序。在你的情况下（addFirst）你得到 UdpProtoBufServerHandler -> ProtobufEncoder -> ... -> ProtobufVarint32FrameDecoder 看起来是错误的，因为你在第一个位置有一个处理程序，在最后有一个解码器。

Answer 2

谢谢大家。我终于想出了办法做到这一点：

public class DatagramServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {

  @Override
  protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
      Message message = SocketProtos.Message.parseFrom(packet.content().nioBuffer());
      ctx.writeAndFlush(packet.retain());
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
      cause.printStackTrace();
      ctx.close();
   }
}

只需使用NioDatagram UDP标准实现，然后将消息解析成byte[]到ProtoBuf消息即可。

Netty + NioDatagram(UDP) + ProtoBuf 问题 - 丢弃入站消息 DatagramPacket

Netty + NioDatagram(UDP) + ProtoBuf issue - Discarded inbound message DatagramPacket

udp

protocol-buffers

netty