本站内所有文档均为中英对照文档，点击中文可以显示英文。此提醒连续关闭5次后，将不再显示。
UDP 服务器
目录
Reactor Netty 提供了易于使用、易于配置的 UdpServer。它隐藏了创建 UDP 服务器所需的大部分 Netty 功能，并加入了 Reactive Streams 背压。
1. 启动与停止
要启动 UDP 服务器，必须创建并配置一个 UdpServer 实例。默认情况下，主机配置为 localhost，端口为 12012。以下示例展示如何创建并启动一个 UDP 服务器：
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create() 1
.bindNow(Duration.ofSeconds(30)); 2
server.onDispose()
.block();
}
}
创建一个 UdpServer 实例，已就绪可供配置。
以阻塞方式启动服务器，并等待其完成初始化。
返回的 Connection 提供了一个简单的服务器 API，其中包括 disposeNow()，它以阻塞方式关闭服务器。
1.1. 主机与端口
要在特定主机和端口上提供服务，您可以对 UDP 服务器应用以下配置：
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.host("localhost") 1
.port(8080) 2
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
配置
UDP服务器主机配置
UDP服务器端口
备注
端口也可通过 PORT 环境变量指定。
2. 预先初始化
默认情况下，UdpServer 资源的初始化是按需进行的。这意味着 bind operation 会吸收初始化和加载以下内容所需的额外时间：
-
事件循环组
-
原生传输库（使用原生传输时）
当您需要预加载这些资源时，可以按如下方式配置 UdpServer：
import io.netty.channel.socket.DatagramPacket;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
UdpServer udpServer =
UdpServer.create()
.handle((in, out) ->
out.sendObject(
in.receiveObject()
.map(o -> {
if (o instanceof DatagramPacket) {
DatagramPacket p = (DatagramPacket) o;
return new DatagramPacket(p.content().retain(), p.sender());
}
else {
return Mono.error(new Exception("Unexpected type of the message: " + o));
}
})));
udpServer.warmup() 1
.block();
Connection server = udpServer.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
初始化并加载事件循环组和原生传输库
3. 写入数据
要向远程对端发送数据，您必须附加一个 I/O 处理器。该 I/O 处理器可访问 UdpOutbound，以便写入数据。以下示例展示如何发送 hello：
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.CharsetUtil;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.handle((in, out) ->
out.sendObject(
in.receiveObject()
.map(o -> {
if (o instanceof DatagramPacket) {
DatagramPacket p = (DatagramPacket) o;
ByteBuf buf = Unpooled.copiedBuffer("hello", CharsetUtil.UTF_8);
return new DatagramPacket(buf, p.sender()); 1
}
else {
return Mono.error(new Exception("Unexpected type of the message: " + o));
}
})))
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
向远程对端发送一个
hello字符串
4. 消费数据
要从远程对端接收数据，您必须附加一个 I/O 处理器。该 I/O 处理器可访问 UdpInbound，以便读取数据。以下示例展示如何消费数据：
import io.netty.channel.socket.DatagramPacket;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.handle((in, out) ->
out.sendObject(
in.receiveObject()
.map(o -> {
if (o instanceof DatagramPacket) {
DatagramPacket p = (DatagramPacket) o;
return new DatagramPacket(p.content().retain(), p.sender()); 1
}
else {
return Mono.error(new Exception("Unexpected type of the message: " + o));
}
})))
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
从远程对端接收数据
5. 生命周期回调
提供以下生命周期回调，供您扩展 UdpServer：
| 回调 | 描述 |
|---|---|
doOnBind | 在服务器通道即将绑定时调用。 |
doOnBound | 在服务器通道已绑定时调用。 |
doOnChannelInit | 在初始化通道时调用。 |
doOnUnbound | 在服务器通道已解绑时调用。 |
以下示例使用 doOnBound 和 doOnChannelInit 回调：
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.logging.LoggingHandler;
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.doOnBound(conn -> conn.addHandlerLast(new LineBasedFrameDecoder(8192))) 1
.doOnChannelInit((observer, channel, remoteAddress) ->
channel.pipeline()
.addFirst(new LoggingHandler("reactor.netty.examples"))) 2
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
服务器通道绑定后，
Netty管道被扩展加入LineBasedFrameDecoder。初始化通道时，
Netty管道被扩展加入LoggingHandler。
6. 连接配置
本节介绍您可在 UDP 级别使用的三种配置：
6.1. 通道选项
默认情况下，UDP 服务器配置了以下选项：
UdpServerBind() {
this.config = new UdpServerConfig(
Collections.singletonMap(ChannelOption.AUTO_READ, false),
() -> new InetSocketAddress(NetUtil.LOCALHOST, DEFAULT_PORT));
如果您需要额外的选项或需要更改当前选项，可以应用以下配置：
import io.netty.channel.ChannelOption;
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
有关 Netty 通道选项的更多信息，请参阅以下链接：
6.2. 线日志
Reactor Netty 提供线日志记录，用于在需要检查对端之间的流量时使用。默认情况下线日志记录为关闭。要启用它，您必须将日志记录器 reactor.netty.udp.UdpServer 的级别设为 DEBUG，并应用以下配置：
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.wiretap(true) 1
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
启用线日志记录
6.2.1. 线日志格式化器
Reactor Netty 支持 3 种不同的格式化器：
/**
* When wire logging is enabled with this format, both events and content will be logged.
* The content will be in hex format.
* <p>Examples:</p>
* <pre>
* {@code
* reactor.netty.http.HttpTests - [d5230a14, L:/0:0:0:0:0:0:0:1:60267 - R:/0:0:0:0:0:0:0:1:60269] REGISTERED
* reactor.netty.http.HttpTests - [d5230a14, L:/0:0:0:0:0:0:0:1:60267 - R:/0:0:0:0:0:0:0:1:60269] ACTIVE
* reactor.netty.http.HttpTests - [d5230a14, L:/0:0:0:0:0:0:0:1:60267 - R:/0:0:0:0:0:0:0:1:60269] READ: 145B
* +-------------------------------------------------+
* | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
* +--------+-------------------------------------------------+----------------+
* |00000000| 50 4f 53 54 20 2f 74 65 73 74 2f 57 6f 72 6c 64 |POST /test/World|
* |00000010| 20 48 54 54 50 2f 31 2e 31 0d 0a 43 6f 6e 74 65 | HTTP/1.1..Conte|
* |00000020| 6e 74 2d 54 79 70 65 3a 20 74 65 78 74 2f 70 6c |nt-Type: text/pl|
* |00000030| 61 69 6e 0d 0a 75 73 65 72 2d 61 67 65 6e 74 3a |ain..user-agent:|
* |00000040| 20 52 65 61 63 74 6f 72 4e 65 74 74 79 2f 64 65 | ReactorNetty/de|
* ...
* reactor.netty.http.HttpTests - [d5230a14, L:/0:0:0:0:0:0:0:1:60267 - R:/0:0:0:0:0:0:0:1:60269] WRITE: 38B
* +-------------------------------------------------+
* | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
* +--------+-------------------------------------------------+----------------+
* |00000000| 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d |HTTP/1.1 200 OK.|
* |00000010| 0a 63 6f 6e 74 65 6e 74 2d 6c 65 6e 67 74 68 3a |.content-length:|
* |00000020| 20 30 0d 0a 0d 0a | 0.... |
* +--------+-------------------------------------------------+----------------+
* }
* </pre>
*/
/**
* When wire logging is enabled with this format, only the events will be logged.
* <p>Examples:</p>
* <pre>
* {@code
* reactor.netty.http.HttpTests - [230d3686, L:/0:0:0:0:0:0:0:1:60241 - R:/0:0:0:0:0:0:0:1:60245] REGISTERED
* reactor.netty.http.HttpTests - [230d3686, L:/0:0:0:0:0:0:0:1:60241 - R:/0:0:0:0:0:0:0:1:60245] ACTIVE
* reactor.netty.http.HttpTests - [230d3686, L:/0:0:0:0:0:0:0:1:60241 - R:/0:0:0:0:0:0:0:1:60245] READ: 145B
* reactor.netty.http.HttpTests - [230d3686, L:/0:0:0:0:0:0:0:1:60241 - R:/0:0:0:0:0:0:0:1:60245] WRITE: 38B
* }
* </pre>
*/
/**
* When wire logging is enabled with this format, both events and content will be logged.
* The content will be in plain text format.
* <p>Examples:</p>
* <pre>
* {@code
* reactor.netty.http.HttpTests - [02c3db6c, L:/0:0:0:0:0:0:0:1:60317 - R:/0:0:0:0:0:0:0:1:60319] REGISTERED
* reactor.netty.http.HttpTests - [02c3db6c, L:/0:0:0:0:0:0:0:1:60317 - R:/0:0:0:0:0:0:0:1:60319] ACTIVE
* reactor.netty.http.HttpTests - [02c3db6c, L:/0:0:0:0:0:0:0:1:60317 - R:/0:0:0:0:0:0:0:1:60319] READ: 145B POST /test/World HTTP/1.1
* Content-Type: text/plain
* user-agent: ReactorNetty/dev
* ...
* reactor.netty.http.HttpTests - [02c3db6c, L:/0:0:0:0:0:0:0:1:60317 - R:/0:0:0:0:0:0:0:1:60319] WRITE: 38B HTTP/1.1 200 OK
* content-length: 0
* }
* </pre>
*/
当您需要更改默认格式化器时，可以按如下方式配置：
import io.netty.handler.logging.LogLevel;
import reactor.netty.Connection;
import reactor.netty.transport.logging.AdvancedByteBufFormat;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.wiretap("logger-name", LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL) 1
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
启用线日志记录，使用 AdvancedByteBufFormat#TEXTUAL 打印内容。
6.3. 事件循环组
默认情况下，Reactor Netty 使用一个“事件循环组”，其中工作线程数等于初始化时运行时可用的处理器数量（但最小值为 4）。该“事件循环组”在一个 JVM 中的所有服务器和客户端之间共享。当您需要不同的配置时，可以使用 LoopResources#create 方法之一。
以下清单展示了事件循环组的默认配置：
/**
* Default worker thread count, fallback to available processor
* (but with a minimum value of 4).
*/
public static final String IO_WORKER_COUNT = "reactor.netty.ioWorkerCount";
/**
* Default selector thread count, fallback to -1 (no selector thread)
* <p><strong>Note:</strong> In most use cases using a worker thread also as a selector thread works well.
* A possible use case for specifying a separate selector thread might be when the worker threads are too busy
* and connections cannot be accepted fast enough.
* <p><strong>Note:</strong> Although more than 1 can be configured as a selector thread count, in reality
* only 1 thread will be used as a selector thread.
*/
public static final String IO_SELECT_COUNT = "reactor.netty.ioSelectCount";
/**
* Default worker thread count for UDP, fallback to available processor
* (but with a minimum value of 4).
*/
public static final String UDP_IO_THREAD_COUNT = "reactor.netty.udp.ioThreadCount";
/**
* Default quiet period that guarantees that the disposal of the underlying LoopResources
* will not happen, fallback to 2 seconds.
*/
public static final String SHUTDOWN_QUIET_PERIOD = "reactor.netty.ioShutdownQuietPeriod";
/**
* Default maximum amount of time to wait until the disposal of the underlying LoopResources
* regardless if a task was submitted during the quiet period, fallback to 15 seconds.
*/
public static final String SHUTDOWN_TIMEOUT = "reactor.netty.ioShutdownTimeout";
/**
* Default value whether the native transport (epoll, io_uring, kqueue) will be preferred,
* fallback it will be preferred when available.
* <p><strong>Note:</strong> On {@code Linux}, {@code Epoll} will be preferred by default.
* If {@code IO_Uring} needs to be configured, a dependency to {@code io.netty:netty-transport-native-io_uring}
* has to be added.
*/
public static final String NATIVE = "reactor.netty.native";
如果您需要更改这些设置，可以应用以下配置：
import reactor.netty.Connection;
import reactor.netty.resources.LoopResources;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
LoopResources loop = LoopResources.create("event-loop", 1, 4, true);
Connection server =
UdpServer.create()
.runOn(loop)
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
6.3.1. 释放事件循环组
- 如果您使用 Reactor Netty 提供的默认
事件循环组，请调用 HttpResources#disposeLoopsAndConnections/#disposeLoopsAndConnectionsLater方法。
备注
释放 HttpResources 意味着每个正在使用它的服务器/客户端都将无法再使用它！
- 如果您使用自定义
LoopResources，请调用 LoopResources#dispose/#disposeLater方法。
备注
释放自定义 LoopResources 意味着每个配置为使用它的服务器/客户端都将无法再使用它！
7. 指标
UDP 服务器支持与 Micrometer 的内置集成。它以 reactor.netty.udp.server 为前缀公开所有指标。
下表提供 UDP 服务器指标的相关信息：
以下额外指标也可用：
ByteBufAllocator 指标
| 指标名称 | 类型 | 描述 |
|---|---|---|
| reactor.netty.bytebuf.allocator.used.heap.memory | Gauge | 堆缓冲区分配器保留的字节数。参见 已用堆内存 |
| reactor.netty.bytebuf.allocator.used.direct.memory | Gauge | 直接缓冲区分配器保留的字节数。参见 已用直接内存 |
| reactor.netty.bytebuf.allocator.heap.arenas | Gauge | 堆 Arena 的数量（使用 PooledByteBufAllocator 时）。参见 堆 Arena |
| reactor.netty.bytebuf.allocator.direct.arenas | Gauge | 直接 Arena 的数量（使用 PooledByteBufAllocator 时）。参见 直接 Arena |
| reactor.netty.bytebuf.allocator.threadlocal.caches | Gauge | 线程本地缓存的数量（使用 PooledByteBufAllocator 时）。参见 线程本地缓存 |
| reactor.netty.bytebuf.allocator.small.cache.size | Gauge | 小型缓存的大小（使用 PooledByteBufAllocator 时）。参见 小型缓存大小 |
| reactor.netty.bytebuf.allocator.normal.cache.size | Gauge | 普通缓存的大小（使用 PooledByteBufAllocator 时）。参见 普通缓存大小 |
| reactor.netty.bytebuf.allocator.chunk.size | Gauge | Arena 的块大小（使用 PooledByteBufAllocator 时）。参见 块大小 |
| reactor.netty.bytebuf.allocator.active.heap.memory | Gauge | 从堆缓冲区池分配的在用缓冲区实际消耗的字节数（使用 PooledByteBufAllocator 时）。参见 活动堆内存 |
| reactor.netty.bytebuf.allocator.active.direct.memory | Gauge | 从直接缓冲区池分配的在用缓冲区实际消耗的字节数（使用 PooledByteBufAllocator 时）。参见 活动直接内存 |
EventLoop 指标
| 指标名称 | 类型 | 描述 |
|---|---|---|
| reactor.netty.eventloop.pending.tasks | Gauge | 事件循环上待处理的任务数。参见 待处理任务 |
以下示例启用该集成：
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.metrics(true) 1
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
}
启用与 Micrometer 的内置集成
当需要将 UDP 服务器指标集成到 Micrometer 以外的系统，或您想提供自己与 Micrometer 的集成时，可以按如下方式提供您自己的指标记录器：
import reactor.netty.Connection;
import reactor.netty.channel.ChannelMetricsRecorder;
import reactor.netty.udp.UdpServer;
import java.net.SocketAddress;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.metrics(true, CustomChannelMetricsRecorder::new) 1
.bindNow(Duration.ofSeconds(30));
server.onDispose()
.block();
}
启用 UDP 服务器指标，并提供 ChannelMetricsRecorder 实现。
8. Unix 域套接字
当使用原生传输时，UdpServer 支持 Unix 域数据报套接字（UDS）。
以下示例展示如何使用 UDS 支持：
import io.netty.channel.unix.DomainDatagramPacket;
import io.netty.channel.unix.DomainSocketAddress;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpServer;
import java.io.File;
public class Application {
public static void main(String[] args) {
Connection server =
UdpServer.create()
.bindAddress(Application::newDomainSocketAddress) 1
.handle((in, out) ->
out.sendObject(
in.receiveObject()
.map(o -> {
if (o instanceof DomainDatagramPacket) {
DomainDatagramPacket p = (DomainDatagramPacket) o;
return new DomainDatagramPacket(p.content().retain(), p.sender());
}
else {
return Mono.error(new Exception("Unexpected type of the message: " + o));
}
})))
.bindNow();
server.onDispose()
.block();
}
指定将使用的
DomainSocketAddress