本站内所有文档均为中英对照文档，点击中文可以显示英文。此提醒连续关闭5次后，将不再显示。
TCP 服务器
目录
Reactor Netty 提供了易于使用、易于配置的 TcpServer。它隐藏了创建 TCP 服务器所需的大部分 Netty 功能，并加入了 Reactive Streams 背压。
1. 启动与停止
要启动 TCP 服务器，您必须创建并配置一个 TcpServer 实例。默认情况下，host 配置为任意本地地址，当调用 bind 操作时，系统会选用一个临时端口。以下示例展示了如何创建并配置 TcpServer 实例：
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create() 1
.bindNow(); 2
server.onDispose()
.block();
}
}
创建一个可用于配置的 TcpServer 实例。
以阻塞方式启动服务器，并等待其完成初始化。
返回的 DisposableServer 提供了简单的服务器 API，其中包括 disposeNow()，它会以阻塞方式关闭服务器。
1.1. 主机与端口
要在特定的 host 和 port 上提供服务，您可以对 TCP 服务器应用以下配置：
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.host("localhost") 1
.port(8080) 2
.bindNow();
server.onDispose()
.block();
}
}
配置
TCP服务器主机配置
TCP服务器端口
要在多个地址上提供服务，在配置好 TcpServer 之后，您可以多次绑定它以获得各自独立的 DisposableServer。所有创建的服务器都会共享诸如 LoopResources 之类的资源，因为它们底层使用同一个配置实例。
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class MultiAddressApplication {
public static void main(String[] args) {
TcpServer tcpServer = TcpServer.create();
DisposableServer server1 = tcpServer
.host("localhost") 1
.port(8080) 2
.bindNow();
DisposableServer server2 = tcpServer
.host("0.0.0.0") 3
.port(8081) 4
.bindNow();
Mono.when(server1.onDispose(), server2.onDispose())
.block();
}
}
配置第一个
TCP服务器主机配置第一个
TCP服务器端口配置第二个
TCP服务器主机配置第二个
TCP服务器端口
2. 立即初始化
默认情况下，TcpServer 资源的初始化按需进行。这意味着 bind operation 会承担初始化和加载所需的额外时间：
-
事件循环组
-
原生传输库（使用原生传输时）
-
安全相关的原生库（
OpenSsl的情形）
当您需要预先加载这些资源时，可以按如下方式配置 TcpServer：
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
TcpServer tcpServer =
TcpServer.create()
.handle((inbound, outbound) -> inbound.receive().then());
tcpServer.warmup() 1
.block();
DisposableServer server = tcpServer.bindNow();
server.onDispose()
.block();
}
}
初始化并加载事件循环组、原生传输库以及安全相关的原生库
3. 写入数据
要向已连接的客户端发送数据，您必须挂载一个 I/O 处理器。该 I/O 处理器可以访问 NettyOutbound，以便写入数据。以下示例展示了如何挂载 I/O 处理器：
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.handle((inbound, outbound) -> outbound.sendString(Mono.just("hello"))) 1
.bindNow();
server.onDispose()
.block();
}
}
向已连接的客户端发送
hello字符串
4. 消费数据
要从已连接的客户端接收数据，您必须挂载一个 I/O 处理器。该 I/O 处理器可以访问 NettyInbound，以便读取数据。以下示例展示了如何使用它：
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.handle((inbound, outbound) -> inbound.receive().then()) 1
.bindNow();
server.onDispose()
.block();
}
}
从已连接的客户端接收数据
5. 生命周期回调
提供以下生命周期回调，以便您扩展 TcpServer：
| 回调 | 说明 |
|---|---|
doOnBind | 在服务器通道即将绑定时调用。 |
doOnBound | 在服务器通道完成绑定时调用。 |
doOnChannelInit | 在初始化通道时调用。 |
doOnConnection | 在远程客户端连接时调用 |
doOnUnbound | 在服务器通道取消绑定时调用。 |
以下示例使用了 doOnConnection 和 doOnChannelInit 回调：
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
import java.util.concurrent.TimeUnit;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.doOnConnection(conn ->
conn.addHandlerFirst(new ReadTimeoutHandler(10, TimeUnit.SECONDS))) 1
.doOnChannelInit((observer, channel, remoteAddress) ->
channel.pipeline()
.addFirst(new LoggingHandler("reactor.netty.examples"))) 2
.bindNow();
server.onDispose()
.block();
}
}
远程客户端连接后，
Netty管道会追加ReadTimeoutHandler。初始化通道时，
Netty管道会追加LoggingHandler。
6. TCP 级配置
本节介绍您可以在 TCP 级别使用的三种配置：
6.1. 设置通道选项
默认情况下，TCP 服务器配置了以下选项：
TcpServerBind() {
Map<ChannelOption<?>, Boolean> childOptions = new HashMap<>(MapUtils.calculateInitialCapacity(2));
childOptions.put(ChannelOption.AUTO_READ, false);
childOptions.put(ChannelOption.TCP_NODELAY, true);
this.config = new TcpServerConfig(
Collections.singletonMap(ChannelOption.SO_REUSEADDR, true),
childOptions,
() -> new InetSocketAddress(DEFAULT_PORT));
}
如果需要额外的选项或要更改当前选项，您可以应用以下配置：
import io.netty.channel.ChannelOption;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
.bindNow();
server.onDispose()
.block();
}
}
您可以在以下链接中找到关于 Netty 通道选项的更多信息：
6.2. 线路日志记录器
当需要检查两个对端之间的流量时，Reactor Netty 提供了线路日志。默认情况下，线路日志处于禁用状态。要启用它，您必须将日志记录器 reactor.netty.tcp.TcpServer 的级别设置为 DEBUG，并应用以下配置：
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.wiretap(true) 1
.bindNow();
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.DisposableServer;
import reactor.netty.tcp.TcpServer;
import reactor.netty.transport.logging.AdvancedByteBufFormat;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.wiretap("logger-name", LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL) 1
.bindNow();
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.DisposableServer;
import reactor.netty.resources.LoopResources;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
LoopResources loop = LoopResources.create("event-loop", 1, 4, true);
DisposableServer server =
TcpServer.create()
.runOn(loop)
.bindNow();
server.onDispose()
.block();
}
}
6.3.1. 释放事件循环组
- 如果您使用 Reactor Netty 提供的默认
事件循环组，请调用 HttpResources#disposeLoopsAndConnections/#disposeLoopsAndConnectionsLater方法。
备注
释放 HttpResources 意味着每个正在使用它的服务器/客户端都将无法再使用它！
- 如果您使用自定义
LoopResources，请调用 LoopResources#dispose/#disposeLater方法。
备注
释放自定义 LoopResources 意味着每个配置为使用它的服务器/客户端都将无法再使用它！
7. SSL 与 TLS
当您需要 SSL 或 TLS 时，可以应用下例所示的配置。默认情况下，如果 OpenSSL 可用，则使用 SslProvider.OPENSSL 作为提供方；否则使用 SslProvider.JDK。可以通过 SslContextBuilder 或设置 -Dio.netty.handler.ssl.noOpenSsl=true 来切换提供方。
以下示例使用了 SslContextBuilder：
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
import reactor.netty.tcp.TcpSslContextSpec;
import java.io.File;
public class Application {
public static void main(String[] args) {
File cert = new File("certificate.crt");
File key = new File("private.key");
TcpSslContextSpec tcpSslContextSpec = TcpSslContextSpec.forServer(cert, key);
DisposableServer server =
TcpServer.create()
.secure(spec -> spec.sslContext(tcpSslContextSpec))
.bindNow();
server.onDispose()
.block();
}
}
7.1. 服务器名称指示
您可以为 TCP 服务器配置多个映射到特定域名的 SslContext。配置 SNI 映射时，可以使用精确域名，也可以使用包含通配符的域名。
以下示例使用了一个包含通配符的域名：
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
import java.io.File;
public class Application {
public static void main(String[] args) throws Exception {
File defaultCert = new File("default_certificate.crt");
File defaultKey = new File("default_private.key");
File testDomainCert = new File("default_certificate.crt");
File testDomainKey = new File("default_private.key");
SslContext defaultSslContext = SslContextBuilder.forServer(defaultCert, defaultKey).build();
SslContext testDomainSslContext = SslContextBuilder.forServer(testDomainCert, testDomainKey).build();
DisposableServer server =
TcpServer.create()
.secure(spec -> spec.sslContext(defaultSslContext)
.addSniMapping("*.test.com",
testDomainSpec -> testDomainSpec.sslContext(testDomainSslContext)))
.bindNow();
server.onDispose()
.block();
}
}
8. 指标
TCP 服务器支持与 Micrometer 的内置集成。它以 reactor.netty.tcp.server 为前缀暴露所有指标。
下表提供了 TCP 服务器指标的相关信息：
| 指标名称 | 类型 | 说明 |
|---|---|---|
| reactor.netty.tcp.server.connections.total | Gauge | 已打开的所有连接数量。请参阅 连接总数 |
| reactor.netty.tcp.server.data.received | DistributionSummary | 接收的数据量，以字节为单位。请参阅 接收数据 |
| reactor.netty.tcp.server.data.sent | DistributionSummary | 发送的数据量，以字节为单位。请参阅 发送数据 |
| reactor.netty.tcp.server.errors | Counter | 发生的错误数量。请参阅 错误数 |
| reactor.netty.tcp.server.tls.handshake.time | Timer | TLS 握手所花费的时间。请参阅 TLS 握手时间 |
以下附加指标同样可用：
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.DisposableServer;
import reactor.netty.tcp.TcpServer;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.metrics(true) 1
.bindNow();
server.onDispose()
.block();
}
}
启用与 Micrometer 的内置集成
当需要将 TCP 服务器指标与 Micrometer 以外的系统集成，或者您想提供自己的与 Micrometer 的集成时，可以按如下方式提供自己的指标记录器：
import reactor.netty.DisposableServer;
import reactor.netty.channel.ChannelMetricsRecorder;
import reactor.netty.tcp.TcpServer;
import java.net.SocketAddress;
import java.time.Duration;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
.metrics(true, CustomChannelMetricsRecorder::new) 1
.bindNow();
server.onDispose()
.block();
}
启用 TCP 服务器指标，并提供 ChannelMetricsRecorder 实现。
9. 追踪
TCP 服务器支持与 Micrometer Tracing 的内置集成。
下表提供了 TCP 服务器跨度的相关信息：
| 上下文名称 | 说明 |
|---|---|
| tls handshake | TLS 握手的相关信息及所花费时间。请参阅 TLS 握手跨度。 |
以下示例启用了该集成。此示例使用 Brave，并将信息上报到 Zipkin。有关 OpenTelemetry 的设置，请参阅 Micrometer Tracing 文档。
import brave.Tracing;
import brave.propagation.StrictCurrentTraceContext;
import brave.sampler.Sampler;
import io.micrometer.tracing.CurrentTraceContext;
import io.micrometer.tracing.Tracer;
import io.micrometer.tracing.brave.bridge.BraveBaggageManager;
import io.micrometer.tracing.brave.bridge.BraveCurrentTraceContext;
import io.micrometer.tracing.brave.bridge.BraveTracer;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.observability.ReactorNettyTracingObservationHandler;
import reactor.netty.tcp.TcpServer;
import zipkin2.reporter.brave.AsyncZipkinSpanHandler;
import zipkin2.reporter.urlconnection.URLConnectionSender;
import static reactor.netty.Metrics.OBSERVATION_REGISTRY;
public class Application {
public static void main(String[] args) {
init(); 1
DisposableServer server =
TcpServer.create()
.metrics(true) 2
.handle((inbound, outbound) -> outbound.sendString(Mono.just("hello")))
.bindNow();
server.onDispose()
.block();
}
/**
* This setup is based on
* <a href="https://micrometer.io/docs/tracing#_micrometer_tracing_brave_setup">Micrometer Tracing Brave Setup</a>.
*/
static void init() {
AsyncZipkinSpanHandler spanHandler = AsyncZipkinSpanHandler
.create(URLConnectionSender.create("http://localhost:9411/api/v2/spans"));
StrictCurrentTraceContext braveCurrentTraceContext = StrictCurrentTraceContext.create();
CurrentTraceContext bridgeContext = new BraveCurrentTraceContext(braveCurrentTraceContext);
Tracing tracing =
Tracing.newBuilder()
.currentTraceContext(braveCurrentTraceContext)
.supportsJoin(false)
.traceId128Bit(true)
.sampler(Sampler.ALWAYS_SAMPLE)
.addSpanHandler(spanHandler)
.localServiceName("reactor-netty-examples")
.build();
brave.Tracer braveTracer = tracing.tracer();
Tracer tracer = new BraveTracer(braveTracer, bridgeContext, new BraveBaggageManager());
OBSERVATION_REGISTRY.observationConfig()
.observationHandler(new ReactorNettyTracingObservationHandler(tracer));
}
}
初始化 Brave、Zipkin 以及 Observation 注册表。
启用与 Micrometer 的内置集成。
在 Zipkin 中的结果如下所示：
9.1. 访问当前观测
Micrometer 项目提供了一个库，可协助在不同类型的上下文机制之间进行上下文传播，例如 ThreadLocal、Reactor Context 等。
以下示例展示了如何在自定义 ChannelHandler 中使用该库：
import brave.Tracing;
import brave.propagation.StrictCurrentTraceContext;
import brave.sampler.Sampler;
import io.micrometer.context.ContextSnapshot;
import io.micrometer.context.ContextSnapshotFactory;
import io.micrometer.tracing.CurrentTraceContext;
import io.micrometer.tracing.Tracer;
import io.micrometer.tracing.brave.bridge.BraveBaggageManager;
import io.micrometer.tracing.brave.bridge.BraveCurrentTraceContext;
import io.micrometer.tracing.brave.bridge.BraveTracer;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.NettyPipeline;
import reactor.netty.observability.ReactorNettyTracingObservationHandler;
import reactor.netty.tcp.TcpServer;
import reactor.netty.tcp.TcpSslContextSpec;
import zipkin2.reporter.brave.AsyncZipkinSpanHandler;
import zipkin2.reporter.urlconnection.URLConnectionSender;
import java.io.File;
import static reactor.netty.Metrics.OBSERVATION_REGISTRY;
public class Application {
public static void main(String[] args) {
init(); 1
File cert = new File("certificate.crt");
File key = new File("private.key");
TcpSslContextSpec tcpSslContextSpec = TcpSslContextSpec.forServer(cert, key);
DisposableServer server =
TcpServer.create()
.metrics(true) 2
.doOnChannelInit((observer, channel, address) -> channel.pipeline().addAfter(
NettyPipeline.SslHandler, "custom-channel-handler", CustomChannelInboundHandler.INSTANCE)) 3
.secure(spec -> spec.sslContext(tcpSslContextSpec))
.handle((inbound, outbound) -> outbound.sendString(Mono.just("hello")))
.bindNow();
server.onDispose()
.block();
}
static final class CustomChannelInboundHandler extends ChannelInboundHandlerAdapter {
static final ChannelHandler INSTANCE = new CustomChannelInboundHandler();
@Override
public void channelActive(ChannelHandlerContext ctx) {
try (ContextSnapshot.Scope scope = ContextSnapshotFactory.builder().build().setThreadLocalsFrom(ctx.channel())) {
System.out.println("Current Observation in Scope: " + OBSERVATION_REGISTRY.getCurrentObservation());
ctx.fireChannelActive();
}
System.out.println("Current Observation: " + OBSERVATION_REGISTRY.getCurrentObservation());
}
@Override
public boolean isSharable() {
return true;
}
}
初始化 Brave、Zipkin 以及 Observation 注册表。
启用与 Micrometer 的内置集成。
使用上下文传播库的自定义
ChannelHandler。此示例仅重写了ChannelInboundHandlerAdapter#channelActive，如有需要，相同逻辑也可用于其余方法。此外，此示例会为给定Channel中存在值的所有ThreadLocal赋值；如果需要其他行为，请查看上下文传播库 API。例如，您可能只想设置部分ThreadLocal值。
备注
当您在某个框架内启用 Reactor Netty 追踪时，可能需要让 Reactor Netty 使用该框架创建的 ObservationRegistry。为此，您需要调用 reactor.netty.Metrics#observationRegistry。您可能还需要使用该框架提供的 API 来配置 Reactor Netty 的 ObservationHandlers。
10. Unix 域套接字
当使用原生传输时，TCP 服务器在所有 Java 版本上都支持 Unix 域套接字（UDS）；当使用 NIO 传输时，在 Java 17 及以上版本上支持。
以下示例展示了如何使用 UDS 支持：
import io.netty.channel.unix.DomainSocketAddress;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;
//import java.net.UnixDomainSocketAddress;
public class Application {
public static void main(String[] args) {
DisposableServer server =
TcpServer.create()
// The configuration below is available only when Epoll/KQueue transport is used
.bindAddress(() -> new DomainSocketAddress("/tmp/test.sock")) 1
// The configuration below is available only when NIO transport is used with Java 17+
//.bindAddress(() -> UnixDomainSocketAddress.of("/tmp/test.sock"))
.bindNow();
server.onDispose()
.block();
}
}
指定将要使用的
DomainSocketAddress