Vert.x 4.5+废弃rxExecuteBlocking的替代方案及RxJava转换方法

从Vert.x 4.5.0开始，Vertx#rxExecuteBlocking以及返回Maybe的executeBlocking签名已被标记为废弃。下面是将executeBlocking返回的Vert.x Future转换为RxJava Maybe或Observable的具体实现方案：

转换为RxJava Maybe #

Maybe适合处理可能返回结果或无结果的场景，对应原rxExecuteBlocking返回Maybe的语义。

方法1：手动封装Maybe #

直接通过RxJava的Maybe.create绑定Vert.x Future的回调逻辑：

import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.reactivex.rxjava3.core.Maybe;

// 执行阻塞任务获取Vert.x Future
Future<String> blockingTaskFuture = vertx.executeBlocking(promise -> {
    // 替换为你的阻塞业务逻辑
    String taskResult = performBlockingOperation();
    // 无结果时可传递null，或直接调用promise.complete()
    promise.complete(taskResult);
});

// 转换为RxJava Maybe
Maybe<String> maybeResult = Maybe.create(emitter -> {
    blockingTaskFuture.onComplete(asyncResult -> {
        if (asyncResult.succeeded()) {
            String result = asyncResult.result();
            if (result != null) {
                emitter.onSuccess(result);
            } else {
                emitter.onComplete();
            }
        } else {
            emitter.onError(asyncResult.cause());
        }
    });
});

方法2：使用Vertx RxHelper工具类 #

Vertx的RxJava集成包提供了RxHelper，可以一键完成转换：

import io.vertx.rxjava3.core.RxHelper;
import io.vertx.core.Future;
import io.reactivex.rxjava3.core.Maybe;

Future<String> blockingTaskFuture = vertx.executeBlocking(promise -> {
    String taskResult = performBlockingOperation();
    promise.complete(taskResult);
});

Maybe<String> maybeResult = RxHelper.maybe(blockingTaskFuture);

转换为RxJava Observable #

Observable适合处理需要发射单个结果的场景（如果需要确保有结果，也可以用Single，对应RxHelper.single()）。

方法1：手动封装Observable #

import io.vertx.core.Future;
import io.reactivex.rxjava3.core.Observable;

Future<String> blockingTaskFuture = vertx.executeBlocking(promise -> {
    String taskResult = performBlockingOperation();
    promise.complete(taskResult);
});

Observable<String> observableResult = Observable.create(emitter -> {
    blockingTaskFuture.onComplete(asyncResult -> {
        if (asyncResult.succeeded()) {
            emitter.onNext(asyncResult.result());
            emitter.onComplete();
        } else {
            emitter.onError(asyncResult.cause());
        }
    });
});

方法2：使用RxHelper工具类 #

import io.vertx.rxjava3.core.RxHelper;
import io.vertx.core.Future;
import io.reactivex.rxjava3.core.Observable;

Future<String> blockingTaskFuture = vertx.executeBlocking(promise -> {
    String taskResult = performBlockingOperation();
    promise.complete(taskResult);
});

Observable<String> observableResult = RxHelper.observable(blockingTaskFuture);

注意事项 #

• 确保项目引入对应版本的Vertx RxJava依赖（以RxJava3为例）：

<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-rx-java3</artifactId>
    <version>你的Vertx版本（需≥4.5.0）</version>
</dependency>

• 若阻塞任务必然返回结果，使用Single（RxHelper.single()）比Observable更贴合语义；若可能无结果，优先选择Maybe。

内容的提问来源于stack exchange，提问作者ccjmne