本站内所有文档均为中英对照文档，点击中文可以显示英文。此提醒连续关闭5次后，将不再显示。
UDP 客户端
目录
Reactor Netty 提供了易于使用、易于配置的 UdpClient。它隐藏了创建 UDP 客户端所需的大部分 Netty 功能，并加入了 Reactive Streams 背压。
1. 连接与断开连接
要将 UDP 客户端连接到给定端点，您必须创建并配置一个 UdpClient 实例。默认情况下，主机配置为 localhost，端口为 12012。以下示例展示如何创建并连接一个 UDP 客户端：
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create() 1
.connectNow(Duration.ofSeconds(30)); 2
connection.onDispose()
.block();
}
}
创建一个 UdpClient 实例，已就绪可供配置。
以阻塞方式连接客户端，并等待其完成初始化。
返回的 Connection 提供了一个简单的连接 API，其中包括 disposeNow()，它以阻塞方式关闭客户端。
1.1. 主机与端口
要连接到特定的 host 和 port，您可以对 UDP 客户端应用以下配置：
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com") 1
.port(80) 2
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
配置此客户端应连接到的
host配置此客户端应连接到的
port
备注
端口也可通过 PORT 环境变量指定。
2. 预先初始化
默认情况下，UdpClient 资源的初始化是按需进行的。这意味着 connect operation 会吸收初始化和加载以下内容所需的额外时间：
-
事件循环组
-
主机名解析器
-
原生传输库（使用原生传输时）
当您需要预加载这些资源时，可以按如下方式配置 UdpClient：
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
UdpClient udpClient = UdpClient.create()
.host("example.com")
.port(80)
.handle((udpInbound, udpOutbound) -> udpOutbound.sendString(Mono.just("hello")));
udpClient.warmup() 1
.block();
Connection connection = udpClient.connectNow(Duration.ofSeconds(30)); 2
connection.onDispose()
.block();
}
}
初始化并加载事件循环组、主机名解析器和原生传输库
主机名解析在连接到远程对端时发生
3. 写入数据
要向给定对端发送数据，您必须附加一个 I/O 处理器。该 I/O 处理器可访问 UdpOutbound，以便写入数据。
以下示例展示如何发送 hello：
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.handle((udpInbound, udpOutbound) -> udpOutbound.sendString(Mono.just("hello"))) 1
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
向远程对端发送
hello字符串。
4. 消费数据
要从给定对端接收数据，您必须附加一个 I/O 处理器。该 I/O 处理器可访问 UdpInbound，以便读取数据。以下示例展示如何消费数据：
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.handle((udpInbound, udpOutbound) -> udpInbound.receive().then()) 1
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
从给定对端接收数据
5. 生命周期回调
提供以下生命周期回调，供您扩展 UdpClient：
| 回调 | 描述 |
|---|---|
doAfterResolve | 在远程地址成功解析后调用。 |
doOnChannelInit | 在初始化通道时调用。 |
doOnConnect | 在通道即将连接时调用。 |
doOnConnected | 在通道连接后调用。 |
doOnDisconnected | 在通道断开连接后调用。 |
doOnResolve | 在远程地址即将被解析时调用。 |
doOnResolveError | 在远程地址未能成功解析时调用。 |
以下示例使用 doOnConnected 和 doOnChannelInit 回调：
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.logging.LoggingHandler;
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.doOnConnected(conn -> conn.addHandlerLast(new LineBasedFrameDecoder(8192))) 1
.doOnChannelInit((observer, channel, remoteAddress) ->
channel.pipeline()
.addFirst(new LoggingHandler("reactor.netty.examples"))) 2
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
通道连接后，
Netty管道被扩展加入LineBasedFrameDecoder。初始化通道时，
Netty管道被扩展加入LoggingHandler。
6. 连接配置
本节介绍您可在 UDP 级别使用的三种配置：
6.1. 通道选项
默认情况下，UDP 客户端配置了以下选项：
UdpClientConnect() {
this.config = new UdpClientConfig(
ConnectionProvider.newConnection(),
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.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
您可通过以下链接进一步了解 Netty 通道选项：
6.2. 线日志
Reactor Netty 提供线日志记录，用于在需要检查对端之间的流量时使用。默认情况下线日志记录为关闭。要启用它，您必须将日志记录器 reactor.netty.udp.UdpClient 的级别设为 DEBUG，并应用以下配置：
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.wiretap(true) 1
.connectNow(Duration.ofSeconds(30));
connection.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.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.wiretap("logger-name", LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL) 1
.connectNow(Duration.ofSeconds(30));
connection.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.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
LoopResources loop = LoopResources.create("event-loop", 1, 4, true);
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.runOn(loop)
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
6.3.1. 释放事件循环组
- 如果您使用 Reactor Netty 提供的默认
事件循环组，请调用 HttpResources#disposeLoopsAndConnections/#disposeLoopsAndConnectionsLater方法。
备注
释放 HttpResources 意味着每个正在使用它的服务器/客户端都将无法再使用它！
- 如果您使用自定义
LoopResources，请调用 LoopResources#dispose/#disposeLater方法。
备注
释放自定义 LoopResources 意味着每个配置为使用它的服务器/客户端都将无法再使用它！
7. 指标
UDP 客户端支持与 Micrometer 的内置集成。它以 reactor.netty.udp.client 为前缀公开所有指标。
下表提供 UDP 客户端指标的相关信息：
| 指标名称 | 类型 | 描述 |
|---|---|---|
| reactor.netty.udp.client.data.received | DistributionSummary | 接收的数据量，以字节为单位。参见 接收数据 |
| reactor.netty.udp.client.data.sent | DistributionSummary | 发送的数据量，以字节为单位。参见 发送数据 |
| reactor.netty.udp.client.errors | Counter | 已发生的错误数。参见 错误数 |
| reactor.netty.udp.client.connect.time | Timer | 连接到远程地址所花费的时间。参见 连接时间 |
| reactor.netty.udp.client.address.resolver | Timer | 解析地址所花费的时间。参见 主机名解析时间 |
以下额外指标也可用：
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.UdpClient;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.metrics(true) 1
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
}
启用与 Micrometer 的内置集成
当需要将 UDP 客户端指标集成到 Micrometer 以外的系统，或您想提供自己与 Micrometer 的集成时，可以按如下方式提供您自己的指标记录器：
import reactor.netty.Connection;
import reactor.netty.channel.ChannelMetricsRecorder;
import reactor.netty.udp.UdpClient;
import java.net.SocketAddress;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.host("example.com")
.port(80)
.metrics(true, CustomChannelMetricsRecorder::new) 1
.connectNow(Duration.ofSeconds(30));
connection.onDispose()
.block();
}
启用 UDP 客户端指标，并提供 ChannelMetricsRecorder 实现。
8. Unix 域套接字
当使用原生传输时，UdpClient 支持 Unix 域数据报套接字（UDS）。
以下示例展示如何使用 UDS 支持：
import io.netty.channel.unix.DomainSocketAddress;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.udp.UdpClient;
import java.io.File;
public class Application {
public static void main(String[] args) {
Connection connection =
UdpClient.create()
.bindAddress(Application::newDomainSocketAddress)
.remoteAddress(() -> new DomainSocketAddress("/tmp/test-server.sock")) 1
.handle((in, out) ->
out.sendString(Mono.just("hello"))
.then(in.receive()
.asString()
.doOnNext(System.out::println)
.then()))
.connectNow();
connection.onDispose()
.block();
}
指定将使用的
DomainSocketAddress