深入理解 `Sinks.Empty＜Void＞` 和 `Mono＜Void＞`：如何触发完成信号并结合 WebSocket 示例

在响应式编程中，Sinks 是 Project Reactor 提供的一个强大工具，用于手动控制数据流的信号发射。Sinks.Empty<Void> 是一种特殊的 Sinks，它不发射任何数据，仅用于表示完成或错误信号。结合 Mono<Void>，它可以用来表示一个异步操作的完成状态。本文将详细分析 Sinks.Empty<Void> 和 Mono<Void> 的行为，并通过示例代码展示其用法，同时结合 WebSocket 建立连接的伪代码，展示其在实际场景中的应用。

1. `Sinks.Empty<Void>` 是什么？

Sinks.Empty<Void> 是 Project Reactor 中的一个 Sinks 类型，专门用于表示一个不发射任何数据的信号源。它的主要特点是：

不发射任何数据（onNext 信号）。
只能发射完成信号（onComplete）或错误信号（onError）。
需要通过显式调用方法（如 tryEmitEmpty()）来触发完成信号。

它的典型使用场景是表示一个异步操作的完成状态，而不需要传递任何数据。

2. `Mono<Void>` 的作用

Mono<Void> 是 Project Reactor 中的一个响应式类型，表示一个最多发射一个元素的异步序列。对于 Mono<Void> 来说：

它不会发射任何数据（onNext 信号）。
它只会发射完成信号（onComplete）或错误信号（onError）。
它通常用于表示一个不需要返回值的异步操作。

通过将 Sinks.Empty<Void> 转换为 Mono<Void>，我们可以将手动控制的信号源与响应式流结合起来。

3. 如何触发完成信号？

Sinks.Empty<Void> 的完成信号需要通过显式调用 tryEmitEmpty() 来触发。以下是关键点：

在调用 tryEmitEmpty() 之前，Sinks.Empty<Void> 处于未完成状态，订阅者会一直等待。
调用 tryEmitEmpty() 后，Sinks.Empty<Void> 会立即发出完成信号，订阅者会收到 onComplete 通知。

4. 示例代码

以下是一个完整的示例，展示了如何使用 Sinks.Empty<Void> 和 Mono<Void> 来触发完成信号：

import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;public class SinksEmptyExample {public static void main(String[] args) {// 1. 创建一个空的 Sinks.Empty<Void>Sinks.Empty<Void> completion = Sinks.empty();// 2. 将其转换为 Mono<Void>Mono<Void> mono = completion.asMono();// 3. 订阅 Mono<Void>mono.subscribe(null, // onNext (不会调用，因为没有数据)error -> System.err.println("Error: " + error), // onError() -> System.out.println("Completed!") // onComplete);// 4. 模拟一个异步操作System.out.println("Starting async operation...");try {Thread.sleep(2000); // 模拟耗时操作} catch (InterruptedException e) {e.printStackTrace();}// 5. 手动触发完成信号System.out.println("Triggering completion...");completion.tryEmitEmpty();// 输出:// Starting async operation...// (等待 2 秒)// Triggering completion...// Completed!}
}

代码解析：

创建 Sinks.Empty<Void>：
- 使用 Sinks.empty() 创建一个空的 Sinks.Empty<Void>。
转换为 Mono<Void>：
- 通过 completion.asMono() 将 Sinks.Empty<Void> 转换为 Mono<Void>。
订阅 Mono<Void>：
- 订阅 Mono<Void>，并定义 onError 和 onComplete 的处理逻辑。
模拟异步操作：
- 使用 Thread.sleep(2000) 模拟一个耗时 2 秒的异步操作。
触发完成信号：
- 调用 completion.tryEmitEmpty() 手动触发完成信号，订阅者会收到 onComplete 通知。

5. 结合 WebSocket 建立连接的伪代码

在实际应用中，Sinks.Empty<Void> 和 Mono<Void> 可以用于表示 WebSocket 连接的建立和关闭。以下是一个伪代码示例，展示如何在 WebSocket 连接建立后触发完成信号：

import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import org.springframework.web.reactive.socket.WebSocketSession;public class WebSocketExample {public Mono<Void> handleWebSocketConnection(WebSocketSession session) {// 1. 创建一个空的 Sinks.Empty<Void>Sinks.Empty<Void> completion = Sinks.empty();// 2. 将其转换为 Mono<Void>Mono<Void> mono = completion.asMono();// 3. 处理 WebSocket 连接session.receive().doOnNext(message -> {// 处理收到的消息System.out.println("Received message: " + message.getPayloadAsText());}).doOnError(error -> {// 处理错误System.err.println("WebSocket error: " + error);completion.tryEmitError(error); // 触发错误信号}).doOnComplete(() -> {// 连接关闭时触发完成信号System.out.println("WebSocket connection closed.");completion.tryEmitEmpty(); // 触发完成信号}).subscribe();// 4. 返回 Mono<Void>，表示 WebSocket 连接的处理结果return mono;}public static void main(String[] args) {WebSocketExample example = new WebSocketExample();// 模拟 WebSocket 连接WebSocketSession session = // 获取 WebSocketSession 的伪代码example.handleWebSocketConnection(session).subscribe(null, // onNext (不会调用，因为没有数据)error -> System.err.println("Error: " + error), // onError() -> System.out.println("WebSocket handling completed!") // onComplete);}
}

伪代码解析：

创建 Sinks.Empty<Void>：
- 使用 Sinks.empty() 创建一个空的 Sinks.Empty<Void>，用于表示 WebSocket 连接的完成状态。
处理 WebSocket 连接：
- 使用 session.receive() 处理 WebSocket 消息。
- 在 doOnNext 中处理收到的消息。
- 在 doOnError 中处理错误，并调用 completion.tryEmitError(error) 触发错误信号。
- 在 doOnComplete 中处理连接关闭，并调用 completion.tryEmitEmpty() 触发完成信号。
返回 Mono<Void>：
- 返回 Mono<Void>，表示 WebSocket 连接的处理结果。
订阅 Mono<Void>：
- 订阅 Mono<Void>，并定义 onError 和 onComplete 的处理逻辑。

6. 关键点总结

Sinks.Empty<Void> 是一个手动控制的信号源，它不会自动发出完成信号，必须通过显式调用 tryEmitEmpty() 来触发。
Mono<Void> 表示一个不发射数据的异步序列，它只会发出完成或错误信号。
WebSocket 连接示例：
- 通过 Sinks.Empty<Void> 和 Mono<Void>，可以灵活地表示 WebSocket 连接的完成状态。
- 在连接关闭或发生错误时，手动触发完成或错误信号。

7. 总结

通过 Sinks.Empty<Void> 和 Mono<Void>，我们可以灵活地控制异步操作的完成信号。结合 WebSocket 示例，展示了如何在实际场景中使用这些工具。关键在于理解 Sinks.Empty<Void> 的初始状态是未完成的，必须通过显式调用 tryEmitEmpty() 或 tryEmitError() 来触发信号。