UDPServer 和加入多播组的任何示例？

Question

刚开始使用 Netty-Reactor，想知道是否有人有任何关于如何使用 UdpServer 加入多播组并处理传入数据包的示例。我还想使用 UdpClient 通过网络发送消息。

当前文档中的示例有点少。

我很困惑我需要编写多少 Netty 自定义对象与 Netty-Reactor 项目对我隐藏的内容..我看到有 UdpInbound 和 UdpOutbound..这些与通道编码器和解码器有什么关系？我还需要那些吗？

我想举一个 UDP 服务器监听多播组的示例，并监听发送到该组的 Google Protobuf 消息。我知道如何在线下进行所有 protobuf 处理..

我只是不确定如何使用 Netty-Reactor 进行实际的在线监听。

我看到 UdpInbound.join() 看起来像我想要的......但在那之后我迷路了。

谁能指点我的例子？

谢谢

更新：

@Violeta - 感谢您的帮助和链接！它对我的帮助更大。

我想我发现了一个错误？或者我可能误解了如何让 IPv4 和 IPv6 在同一台服务器上工作。我注意到您在示例代码中使用了.runOn(resources, InternetProtocolFamily.IPv4)。

如果没有此代码，当我有一个同时分配了 IPv4 和 IPv6 地址的 InetInterface 时，我的绑定会失败。

我希望在同一台服务器和所有接口上同时支持 IPv4/IPv6，因为代码将在多宿主计算机上运行。

除了使用.runOn 方法之外，我没有看到在连接上指定InternetProtocolFamily 的方法。

我收到以下错误：

IFace: name:en0 (en0) Address class: /fe80:0:0:0:1ce8:a828:3c15:4f4b%en0 is Inet4Address = false
IFace: name:en0 (en0) Address class: /192.168.1.151 is Inet4Address = true
09:42:35.615 [udp-nio-1] INFO server.Application - Joining iFace name:en0 (en0) to Multicast Group /224.0.0.224 
09:42:35.621 [udp-nio-1] ERROR reactor.Flux.ConcatArray.1 - onError(java.lang.IllegalArgumentException: IPv6 socket cannot join IPv4 multicast group)
09:42:35.623 [udp-nio-1] ERROR reactor.Flux.ConcatArray.1 - 
java.lang.IllegalArgumentException: IPv6 socket cannot join IPv4 multicast group
    at sun.nio.ch.DatagramChannelImpl.innerJoin(DatagramChannelImpl.java:814)
    at sun.nio.ch.DatagramChannelImpl.join(DatagramChannelImpl.java:900)
    at io.netty.channel.socket.nio.NioDatagramChannel.joinGroup(NioDatagramChannel.java:414)
    at io.netty.channel.socket.nio.NioDatagramChannel.joinGroup(NioDatagramChannel.java:391)
    at io.netty.channel.socket.nio.NioDatagramChannel.joinGroup(NioDatagramChannel.java:384)
    at reactor.netty.udp.UdpOperations.join(UdpOperations.java:65)
    at server.Application.lambda$null$2(Application.java:45)

Answer 1

UdpInbound.join()返回Mono<Void>，加入群后发布者就完成了。 之后您就可以使用UdpInbound.receive() 开始接收包裹了。

in.join(...)
  .thenMany(in.receive()
              ...)
...

有关一些示例，请参阅 here。

您可以使用handler 方法来处理I/O（如链接中所示），或者您可以绑定，当您收到Connection 时，您可以调用Connection.inbound()，然后您可以加入并开始接收包裹。

关于解码器/编码器，我们鼓励您使用 Reactor Core 中的运算符来转换传入的包。如果您无法通过该方法实现此目的，您可以随时添加标准 Netty 处理程序，例如使用 doOnBound(conn -> conn.addHandler...)