在使用 SmallRye Mutiny 进行异步事件处理时，你可能会遇到订阅者没有接收到任何事件的问题，即使你已经使用了 runSubscriptionOn 方法将处理逻辑放在了单独的线程池中执行。这通常是由于 Reactive Streams 规范中的背压机制导致的。
Reactive Streams 是一种用于处理异步数据流的标准，它内置了背压机制，允许消费者控制生产者发送数据的速率，避免消费者被大量数据压垮。在使用 Mutiny 时，你需要理解并正确处理背压，才能确保异步事件处理流程的正常运行。
解决背压问题：使用 Subscription 对象
在 Reactive Streams 中，Subscription 对象代表了发布者和订阅者之间的连接。订阅者需要在 onSubscribe 方法中保存 Subscription 对象，并通过调用其 request(long) 方法来请求数据。
以下是一个示例，展示了如何使用 Subscription 对象来解决异步事件处理中的背压问题：
import io.smallrye.mutiny.Multi;
import org.reactivestreams.Subscription;
import org.reactivestreams.Subscriber;
import java.util.concurrent.Executor;
public class MutinySubscriptionExample {
private final Executor managedExecutor;
public MutinySubscriptionExample(Executor managedExecutor) {
this.managedExecutor = managedExecutor;
}
public void processEvents(Multi events) {
events
.runSubscriptionOn(managedExecutor)
.subscribe()
.withSubscriber(
new Subscriber() {
private Subscription subscription;
@Override
public void onSubscribe(Subscription s) {
System.out.println("OnSubscription Method");
System.out.println("ON SUBS END");
subscription = s;
// 请求第一个事件
subscription.request(1);
}
@Override
public void onNext(String event) {
System.out.println("On Next Method: " + event);
// 处理完一个事件后，请求下一个事件
subscription.request(1);
}
@Override
public void onError(Throwable t) {
System.out.println("OnError Method: " + t.getMessage());
}
@Override
public void onComplete() {
System.out.println("On Complete Method");
}
});
}
public static void main(String[] args) throws InterruptedException {
// 创建一个简单的 Multi 对象
Multi events = Multi.createFrom().items("Event 1", "Event 2", "Event 3");
// 创建一个模拟的 Executor
Executor executor = Runnable::run; // 直接在当前线程执行
// 创建 MutinySubscriptionExample 实例并处理事件
MutinySubscriptionExample example = new MutinySubscriptionExample(executor);
example.processEvents(events);
// 为了确保异步执行完成，等待一段时间
Thread.sleep(100);
}
} 在这个例子中，我们在 onSubscribe 方法中保存了 Subscription 对象，并调用 subscription.request(1) 请求第一个事件。然后在 onNext 方法中，处理完一个事件后，再次调用 subscription.request(1) 请求下一个事件。这样，订阅者就可以逐步接收并处理事件，避免了背压问题。
注意事项：
- request(long) 方法的参数表示请求的事件数量。你可以根据实际情况调整这个值。例如，如果你可以一次性处理多个事件，可以请求更多的事件。
- 如果发布者发送的事件数量超过了订阅者请求的数量，剩余的事件将被缓存，直到订阅者再次请求。
使用 Mutiny 的简洁 API
Mutiny 提供了一组更简洁的 API，可以更方便地处理订阅、事件、错误和完成事件。使用这些 API，你可以避免直接操作 Subscription 对象，使代码更易读和维护。
以下是一个使用 Mutiny 简洁 API 的示例：
import io.smallrye.mutiny.Multi;
import java.util.concurrent.Executor;
public class MutinySimplifiedExample {
private final Executor managedExecutor;
public MutinySimplifiedExample(Executor managedExecutor) {
this.managedExecutor = managedExecutor;
}
public void processEvents(Multi events) {
events
.runSubscriptionOn(managedExecutor)
.onSubscription()
.invoke(() -> {
System.out.println("OnSubscription Method");
System.out.println("ON SUBS END");
})
.onItem()
.invoke(event -> System.out.println("On Next Method: " + event))
.onFailure()
.invoke(t -> System.out.println("OnError Method: " + t.getMessage()))
.onCompletion()
.invoke(() -> System.out.println("On Complete Method"))
.subscribe()
.with(value -> {}); // 必须提供一个消费者，即使它什么也不做
}
public static void main(String[] args) throws InterruptedException {
// 创建一个简单的 Multi 对象
Multi events = Multi.createFrom().items("Event 1", "Event 2", "Event 3");
// 创建一个模拟的 Executor
Executor executor = Runnable::run; // 直接在当前线程执行
// 创建 MutinySimplifiedExample 实例并处理事件
MutinySimplifiedExample example = new MutinySimplifiedExample(executor);
example.processEvents(events);
// 为了确保异步执行完成，等待一段时间
Thread.sleep(100);
}
} 在这个例子中，我们使用了 onSubscription、onItem、onFailure 和 onCompletion 方法来分别处理订阅、事件、错误和完成事件。subscribe().with(value -> {}) 必须提供一个消费者，即使它什么也不做，否则订阅不会启动。
总结：
在使用 SmallRye Mutiny 进行异步事件处理时，理解和处理 Reactive Streams 的背压机制至关重要。你可以通过 Subscription 对象和 request(long) 方法手动控制数据的请求，也可以使用 Mutiny 提供的更简洁的 API 来简化代码。选择哪种方式取决于你的具体需求和偏好。无论选择哪种方式，都要确保订阅者能够及时请求数据，避免阻塞事件流。
