跳到主要内容

QUIC 客户端

Doubao 中英对照 QUIC Client

目录

Reactor Netty 提供了易于使用且易于配置的 QuicClient。它隐藏了创建 QUIC 客户端并添加 Reactive Streams 背压所需的大部分 Netty 功能。

1. 连接与断开​

要将 QUIC 客户端连接到给定端点，您必须创建并配置一个 QuicClient 实例。默认情况下，主机配置为 localhost，端口为 12012。以下示例展示了如何创建并连接一个 QUIC 客户端：

import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

import java.time.Duration;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				QuicClient.create()                            1
				          .connectNow(Duration.ofSeconds(30)); 2

		connection.onDispose()
		          .block();
	}
}

  • 创建一个可供配置的 QuicClient 实例。

  • 以阻塞方式连接客户端，并等待其完成初始化。

返回的 QuicConnection 提供了简单的连接 API，包括 disposeNow()，它以阻塞方式关闭客户端。

1.1. 主机与端口​

要连接到特定的 host 和 port，您可以对 QUIC 客户端应用以下配置：

import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

import java.net.InetSocketAddress;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				QuicClient.create()
				          .remoteAddress(() -> new InetSocketAddress("127.0.0.1", 8080)) 1
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

  • 配置此客户端应连接的 host 和 port

备注

端口也可以通过 QUIC_PORT 环境变量来指定。

2. 安全连接​

要为 QUIC 客户端配置 SSL，您必须应用以下配置：

import io.netty.handler.codec.quic.QuicSslContext;
import io.netty.handler.codec.quic.QuicSslContextBuilder;
import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

public class Application {

	public static void main(String[] args) {
		QuicSslContext clientCtx =
				QuicSslContextBuilder.forClient()
				                     .applicationProtocols("http/1.1")
				                     .build();

		Connection connection =
				QuicClient.create()
				          .secure(clientCtx) 1
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

  • 配置用于安全连接的 SSL 上下文

3. 初始设置​

QUIC 允许为连接配置各种设置。您可以应用以下配置：

import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				QuicClient.create()
				          .initialSettings(spec ->                           1
				              spec.maxData(10000000)                         2
				                  .maxStreamDataBidirectionalLocal(1000000)  3
				                  .maxStreamDataBidirectionalRemote(1000000) 4
				                  .maxStreamDataUnidirectional(1000000)      5
				                  .maxStreamsBidirectional(100)              6
				                  .maxStreamsUnidirectional(100))            7
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

  • 配置 QUIC 连接的初始设置

  • 设置连接上可发送的最大数据量

  • 设置由本端端点发起的双向流上可发送的最大数据量

  • 设置由远程端点发起的双向流上可发送的最大数据量

  • 设置单向流上可发送的最大数据量

  • 设置并发双向流的最大数量

  • 设置并发单向流的最大数量

4. 打开流​

要向远程对端打开一条流，您可以使用以下方式：

import io.netty.handler.codec.quic.QuicSslContext;
import io.netty.handler.codec.quic.QuicSslContextBuilder;
import reactor.core.publisher.Mono;
import reactor.netty.quic.QuicClient;
import reactor.netty.quic.QuicConnection;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;

public class Application {

	public static void main(String[] args) throws InterruptedException {
		QuicSslContext clientCtx =
				QuicSslContextBuilder.forClient()
				                     .applicationProtocols("http/1.1")
				                     .build();

		QuicConnection connection =
				QuicClient.create()
				          .bindAddress(() -> new InetSocketAddress(0))
				          .remoteAddress(() -> new InetSocketAddress("127.0.0.1", 8080))
				          .secure(clientCtx)
				          .idleTimeout(Duration.ofSeconds(5))
				          .initialSettings(spec ->
				              spec.maxData(10000000)
				                  .maxStreamDataBidirectionalLocal(1000000))
				          .connectNow();

		CountDownLatch latch = new CountDownLatch(1);
		connection.createStream((in, out) ->             1
		              out.sendString(Mono.just("Hello")) 2
		                 .then(in.receive()              3
		                         .asString()
		                         .doOnNext(s -> {
		                             System.out.println("CLIENT RECEIVED: " + s);
		                             latch.countDown();
		                         })
		                         .then()))
		          .subscribe();

		latch.await();

		connection.onDispose()
		          .block();
	}
}

  • 创建一条新的双向流

  • 向远程对端发送数据

  • 从远程对端接收数据

5. 生命周期回调​

提供了以下生命周期回调，供您扩展 QuicClient：

回调说明
doOnBind在客户端即将绑定时调用。
doOnBound在客户端绑定完成后调用。
doOnChannelInit在初始化通道时调用。
doOnConnect在客户端即将连接到远程地址时调用。
doOnConnected在客户端连接完成后调用。
doOnDisconnected在客户端断开连接后调用。
doOnUnbound在客户端解除绑定时调用。

以下示例使用了 doOnConnected 回调：

import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

public class Application {

	public static void main(String[] args) throws InterruptedException {
		Connection connection =
				QuicClient.create()
				          .doOnConnected(quicConnection -> System.out.println("Connected!")) 1
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

  • 在客户端连接完成后调用

6. 连接配置​

本节介绍了三种可在 QUIC 客户端级别使用的配置：

6.1. 通道选项​

如果您需要额外的选项，或需要更改当前选项，可以应用以下配置：

import io.netty.channel.ChannelOption;
import io.netty.handler.codec.quic.QLogConfiguration;
import io.netty.handler.codec.quic.QuicChannelOption;
import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

public class Application {

	public static void main(String[] args) throws InterruptedException {
		Connection connection =
				QuicClient.create()
				          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)             1
				          .option(QuicChannelOption.QLOG,
				              new QLogConfiguration("path", "logTitle", "logDescription")) 2
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

  • 配置连接超时

  • 为 QUIC 连接启用 QLOG

您可以通过以下链接了解更多关于 Netty 通道选项的信息：

6.2. 线路日志记录器​

当需要检查对等方之间的流量时，Reactor Netty 提供了线路日志记录功能。默认情况下，线路日志记录是禁用的。要启用它，您必须将日志记录器 reactor.netty.quic.QuicClient 的级别设置为 DEBUG，并应用以下配置：

import reactor.netty.Connection;
import reactor.netty.quic.QuicClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				QuicClient.create()
				          .wiretap(true) 1
				          .connectNow();

		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.quic.QuicClient;
import reactor.netty.transport.logging.AdvancedByteBufFormat;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				QuicClient.create()
				          .wiretap("logger-name", LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL) 1
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

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.quic.QuicClient;
import reactor.netty.resources.LoopResources;

public class Application {

	public static void main(String[] args) {
		LoopResources loop = LoopResources.create("event-loop", 1, 4, true);
		Connection connection =
				QuicClient.create()
				          .runOn(loop)
				          .connectNow();

		connection.onDispose()
		          .block();
	}
}

6.3.1. 释放事件循环组​

  • 如果您使用 Reactor Netty 提供的默认 Event Loop Group，请调用 HttpResources#disposeLoopsAndConnections/#disposeLoopsAndConnectionsLater 方法。
备注

释放 HttpResources 意味着正在使用它的每个服务器/客户端都将无法再使用它！

  • 如果您使用自定义的 LoopResources，请调用 LoopResources#dispose/#disposeLater 方法。
备注

释放自定义的 LoopResources 意味着每个配置为使用它的服务器/客户端都将无法再使用它！

7. 指标​

当需要 QUIC 客户端指标以与 Micrometer 等系统集成时，您可以提供自己的指标记录器，如下所示：

import reactor.netty.Connection;
import reactor.netty.channel.ChannelMetricsRecorder;
import reactor.netty.quic.QuicClient;

import java.net.SocketAddress;
import java.time.Duration;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				QuicClient.create()
				          .metrics(true, CustomChannelMetricsRecorder::new) 1
				          .connectNow();

		connection.onDispose()
		          .block();
	}