深入理解 `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!
    }
}

代码解析：

1. 创建 Sinks.Empty<Void>：

   • 使用 Sinks.empty() 创建一个空的 Sinks.Empty<Void>。

2. 转换为 Mono<Void>：

   • 通过 completion.asMono() 将 Sinks.Empty<Void> 转换为 Mono<Void>。

3. 订阅 Mono<Void>：

   • 订阅 Mono<Void>，并定义 onError 和 onComplete 的处理逻辑。

4. 模拟异步操作：

   • 使用 Thread.sleep(2000) 模拟一个耗时 2 秒的异步操作。

5. 触发完成信号：

   • 调用 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
        );
    }
}

伪代码解析：

1. 创建 Sinks.Empty<Void>：

   • 使用 Sinks.empty() 创建一个空的 Sinks.Empty<Void>，用于表示 WebSocket 连接的完成状态。

2. 处理 WebSocket 连接：

   • 使用 session.receive() 处理 WebSocket 消息。
   • 在 doOnNext 中处理收到的消息。
   • 在 doOnError 中处理错误，并调用 completion.tryEmitError(error) 触发错误信号。
   • 在 doOnComplete 中处理连接关闭，并调用 completion.tryEmitEmpty() 触发完成信号。

3. 返回 Mono<Void>：

   • 返回 Mono<Void>，表示 WebSocket 连接的处理结果。

4. 订阅 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() 来触发信号。