跳到主要内容

QUIC 服务器

Doubao 中英对照 QUIC Server

目录

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

1. 启动与停止​

要启动一个 QUIC 服务器，您必须创建并配置一个 QuicServer 实例。默认情况下，host 配置为任意本地地址，并且在调用 bind 操作时系统会自动选择一个临时端口。以下示例展示了如何创建并启动一个 QUIC 服务器：

import reactor.netty.Connection;
import reactor.netty.quic.QuicServer;

import java.time.Duration;

public class Application {

	public static void main(String[] args) {
		Connection server =
				QuicServer.create()                         1
				          .bindNow(Duration.ofSeconds(30)); 2

		server.onDispose()
		      .block();
	}
}

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

  • 以阻塞方式启动服务器，并等待其完成初始化。

返回的 Connection 提供了简单的服务器 API，包括 disposeNow()，它以阻塞方式关闭服务器。

1.1. 主机与端口​

要绑定到特定的 host 和 port，您可以对 QUIC 服务器应用以下配置：

import reactor.netty.Connection;
import reactor.netty.quic.QuicServer;

public class Application {

	public static void main(String[] args) {
		Connection server =
				QuicServer.create()
				          .host("localhost") 1
				          .port(8080)        2
				          .bindNow();

		server.onDispose()
		      .block();
	}
}

  • 配置此服务器应绑定的 host

  • 配置此服务器应绑定的 port

2. 安全连接​

要为 QUIC 服务器配置 SSL，您必须应用以下配置：

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

import java.io.File;

public class Application {

	public static void main(String[] args) {
		File cert = new File("an X.509 certificate chain file in PEM format");
		File key = new File("a PKCS#8 private key file in PEM format");

		QuicSslContext serverCtx =
				QuicSslContextBuilder.forServer(key, null, cert)
				                     .applicationProtocols("http/1.1")
				                     .build();

		Connection server =
				QuicServer.create()
				          .tokenHandler(InsecureQuicTokenHandler.INSTANCE)
				          .secure(serverCtx) 1
				          .bindNow();

		server.onDispose()
		      .block();
	}
}

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

3. 初始设置​

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

import reactor.netty.Connection;
import reactor.netty.quic.QuicServer;

public class Application {

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

		server.onDispose()
		      .block();
	}
}

  • 配置 QUIC 连接的初始设置

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

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

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

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

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

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

4. 处理流​

要处理传入的流，您可以使用以下方式：

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

import java.io.File;
import java.time.Duration;

public class Application {

	public static void main(String[] args) throws Exception {
		File cert = new File("an X.509 certificate chain file in PEM format");
		File key = new File("a PKCS#8 private key file in PEM format");

		QuicSslContext serverCtx =
				QuicSslContextBuilder.forServer(key, null, cert)
				                     .applicationProtocols("http/1.1")
				                     .build();

		Connection server =
				QuicServer.create()
				          .host("127.0.0.1")
				          .port(8080)
				          .tokenHandler(InsecureQuicTokenHandler.INSTANCE)
				          .secure(serverCtx)
				          .idleTimeout(Duration.ofSeconds(5))
				          .initialSettings(spec ->
				              spec.maxData(10000000)
				                  .maxStreamDataBidirectionalRemote(1000000)
				                  .maxStreamsBidirectional(100))
				          .handleStream((in, out) ->           1
				              out.send(in.receive().retain())) 2
				          .bindNow();

		server.onDispose()
		      .block();
	}
}

  • 在打开新的 QUIC 流时调用

  • 为传入的流设置一个处理器

5. 生命周期回调​

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

回调说明
doOnBind在服务器即将绑定时调用。
doOnBound在服务器绑定完成后调用。
doOnChannelInit在初始化通道时调用。
doOnConnection在建立新的 QUIC 连接时调用。
doOnUnbound在服务器解除绑定时调用。

以下示例使用了 doOnBound 和 doOnConnection 回调：

import reactor.netty.Connection;
import reactor.netty.quic.QuicServer;

public class Application {

	public static void main(String[] args) throws Exception {
		Connection server =
				QuicServer.create()
				          .doOnBound(connection -> System.out.println("Bound!"))                   1
				          .doOnConnection(quicConnection -> System.out.println("New connection!")) 2
				          .bindNow();

		server.onDispose()
		      .block();
	}
}

  • 在服务器绑定完成后调用

  • 在建立新的 QUIC 连接时调用

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.QuicServer;

public class Application {

	public static void main(String[] args) throws Exception {
		Connection server =
				QuicServer.create()
				          .option(ChannelOption.SO_BACKLOG, 1024)                          1
				          .option(QuicChannelOption.QLOG,
				              new QLogConfiguration("path", "logTitle", "logDescription")) 2
				          .bindNow();

		server.onDispose()
		      .block();
	}
}

  • 配置积压队列大小

  • 为 QUIC 连接启用 QLOG

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

6.2. 线路日志记录器​

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

import reactor.netty.Connection;
import reactor.netty.quic.QuicServer;

public class Application {

	public static void main(String[] args) throws Exception {
		Connection server =
				QuicServer.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.Connection;
import reactor.netty.quic.QuicServer;
import reactor.netty.transport.logging.AdvancedByteBufFormat;

public class Application {

	public static void main(String[] args) throws Exception {
		Connection server =
				QuicServer.create()
				          .wiretap("logger-name", LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL) 1
				          .bindNow();

		server.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.QuicServer;
import reactor.netty.resources.LoopResources;

public class Application {

	public static void main(String[] args) throws Exception {
		LoopResources loop = LoopResources.create("event-loop", 1, 4, true);
		Connection server =
				QuicServer.create()
				          .runOn(loop)
				          .bindNow();

		server.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.QuicServer;

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

public class Application {

	public static void main(String[] args) throws Exception {
		Connection server =
				QuicServer.create()
				          .metrics(true, CustomChannelMetricsRecorder::new) 1
				          .bindNow();

		server.onDispose()
		      .block();
	}