响应式编程扩展
API 文档
你当前正在浏览历史版本 Vert.x 4.1.8 的文档。 访问本页面的 最新版本。

Vert.x RxJava

Preview

面向 RxJava3 的 Vert.x API

RxJava 是 Java 虚拟机上极受欢迎的响应式扩展库，利用可观测序列（Observable Sequences）来编排异步与基于事件的程序。

Vert.x 与 RxJava 自然融合，凡是能使用数据流或异步结果的地方，都可以无缝使用 RxJava。

引入并使用面向 RxJava3 的 Vert.x API

要使用面向 RxJava3 的 Vert.x API，请在项目构建描述文件的 dependencies 配置节中添加以下依赖：

  • Maven（在 pom.xml 中）：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-rx-java3</artifactId>
  <version>5.0.12</version>
</dependency>

  • Gradle（在 build.gradle 文件中）：

compile 'io.vertx:vertx-rx-java3:5.0.12'

在 Vert.x 中使用 RxJava 3 API 有两种主要方式：

ReadStream（读流）支持

RxJava Flowable 与 Vert.x 的 ReadStream 类契合得天衣无缝：两者均表示元素流。

FlowableHelper.toFlowable 静态方法可以将 Vert.x 读流转换为 Flowable：

FileSystem fileSystem = vertx.fileSystem();
fileSystem.open("/data.txt", new OpenOptions()).onComplete(result -> {
  AsyncFile file = result.result();
  Flowable<Buffer> observable = FlowableHelper.toFlowable(file);
  observable.forEach(data -> System.out.println("Read data: " + data.toString("UTF-8")));
});

Rx 化 的 Vert.x API 直接在 ReadStream 上提供了 toFlowable 方法：

FileSystem fs = vertx.fileSystem();
fs.rxOpen("/data.txt", new OpenOptions())
  .flatMapPublisher(file -> file.toFlowable())
  .subscribe(data -> System.out.println("Read data: " + data.toString("UTF-8")));

此类 Flowable 属于热流（Hot Flowable），即无论是否存在订阅，它们都可能会产生通知，因为 ReadStream 取决于具体实现，可能会自发地产生事件：

在订阅时，适配器会调用 handler 设置自己的处理器。

某些 ReadStream 实现在此调用后才开始发射事件，另一些则无论是否设置了处理器都会持续发射事件：

  • AsyncFile 在设置处理器后才产生缓冲区事件；

  • HttpServerRequest 的事件产生独立于处理器（即若未设置处理器，数据缓冲区可能会丢失）。

在这两种情况下，在同一调用链中订阅 Flowable 都是安全的，因为 EventLoop 或 Worker Verticle 不会被并发调用，订阅总是发生在处理器开始发射数据之前。

当你需要延迟订阅时，你需要先 pause 暂停 ReadStream，随后再 resume 恢复它，这与操作原生 ReadStream 的方式一致：

server.requestHandler(request -> {
  if (request.method() == HttpMethod.POST) {

    // 暂停接收缓冲区数据
    request.pause();

    checkAuth(res -> {

      // 恢复接收缓冲区数据
      request.resume();

      if (res.succeeded()) {
        Flowable<Buffer> flowable = request.toFlowable();
        flowable.subscribe(buff -> {
          // 处理数据缓冲区
        });
      }
    });
  }
});

同样，也可以将现有的 Flowable 转换为 Vert.x ReadStream。

FlowableHelper.toReadStream 静态方法可以将 Flowable 转换为 Vert.x 读流：

Flowable<Buffer> observable = getFlowable();
ReadStream<Buffer> readStream = FlowableHelper.toReadStream(observable);
readStream.pipeTo(response);

WriteStream（写流）支持

WriteStream 与 org.reactivestreams.Subscriber 类似，负责消费数据项，并在无法跟上生产速率时与生产者协同配合，以避免积压无限扩大。

Vert.x 提供了 WriteStreamSubscriber 适配器，你可以用它将 Flowable 中的元素写入任何 WriteStream：

将缓冲区发送至 HTTP 服务器响应
response.setChunked(true);
WriteStreamSubscriber<io.vertx.core.buffer.Buffer> subscriber = io.vertx.rxjava3.RxHelper.toSubscriber(response);
flowable.subscribe(subscriber);
Vert.x 还为无背压支持的 io.reactivex.Observable 提供了 io.vertx.rxjava3.WriteStreamObserver 适配器。区别在于该适配器即使在写入速率跟不上生产速率时，也会持续向 WriteStream 推送元素。

如果你使用 Rx 化 的 Vert.x API 进行开发，WriteStream 实现直接提供了 toSubscriber 便捷方法。上述示例因而变得更加清晰：

response.setChunked(true);
flowable.subscribe(response.toSubscriber());
当 Flowable 成功终止时，适配器会自动调用 end 方法。
该适配器会设置 WriteStream 的 drain 抽空处理器与 exception 异常处理器，因此在订阅后请勿自行覆盖设置它们。

WriteStreamSubscriber 适配器能够在以下时机触发回调：

  • Flowable 发生错误终止时；

  • WriteStream 发生故障时（例如 HTTP 连接断开或文件系统已满）；

  • WriteStream 正常结束时（即所有写入完成且文件已关闭）；

  • WriteStream 异常结束时（即所有写入完成但在关闭文件时发生错误）。

这有助于构建更加健壮的程序架构，并在数据流处理完成后调度执行后续任务：

response.setChunked(true);

WriteStreamSubscriber<Buffer> subscriber = response.toSubscriber();

subscriber.onError(throwable -> {
  if (!response.headWritten() && response.closed()) {
    response.setStatusCode(500).end("oops");
  } else {
    // 记录错误日志
  }
});

subscriber.onWriteStreamError(throwable -> {
  // 记录写流错误日志
});

subscriber.onWriteStreamEnd(() -> {
  // 向审计系统记录事务完成日志...
});

flowable.subscribe(subscriber);
若 WriteStream 发生异常，适配器会自动取消 org.reactivestreams.Subscription 订阅。

异步结果支持

你可以从现有的 Vert.x Handler<AsyncResult<T>> 创建 RxJava Observer 并对其进行订阅：

Handler<AsyncResult<String>> handler = getHandler();

// 订阅 Single
Single.just("hello").subscribe(SingleHelper.toObserver(handler));
Handler<AsyncResult<String>> handler = getHandler();

// 订阅 Maybe
Maybe.just("hello").subscribe(MaybeHelper.toObserver(handler));
Handler<AsyncResult<Void>> handler = getHandler();

// 订阅 Completable
Completable.complete().subscribe(CompletableHelper.toObserver(handler));

Rx 化 的 Vert.x API 为每个此类异步方法都生成了带有 rx 前缀的衍生方法，返回 RxJava Single、Maybe 或 Completable：

Single<HttpServer> single = vertx
  .createHttpServer()
  .rxListen(1234, "localhost");

// 订阅以绑定端口并启动服务
single.
    subscribe(
        server -> {
          // 服务器正在监听
        },
        failure -> {
          // 服务器启动失败
        }
    );

此类 Single 属于冷流（Cold Single），对应的底层 API 方法只会在被订阅时才真正被调用。

Maybe 可以产生单个结果或不产生结果：

DnsClient client = vertx.createDnsClient(dnsPort, dnsHost);

// 获取一个在订阅时才真正执行反向解析的 Maybe
Maybe<String> maybe = client.rxReverseLookup(ipAddress);

// 订阅以执行反向查找
maybe.
  subscribe(
    name -> {
      // 成功解析出名称
    },
    failure -> {
      // 解析失败
    },
    () -> {
      // 解析完成但未产生结果
    }
  );

Completable 通常映射到 Handler<AsyncResult<Void>>：

Completable single = server.rxClose();

// 订阅以关闭服务器
single.
  subscribe(
    () -> {
      // 服务器已成功关闭
    },
    failure -> {
      // 服务器关闭时发生异常
    }
  );

如果你无法使用 Vert.x Rx 化 API，或者你拥有基于回调的自定义异步方法，Vert.x 提供了相应的转换适配器：
适配 Vert.x 核心的 executeBlocking 方法
Maybe<String> maybe = MaybeHelper.toMaybe(handler -> {
  vertx.executeBlocking(() -> invokeBlocking()).onComplete(handler);
});

调度器（Scheduler）支持

响应式扩展库有时需要调度动作，例如 Flowable#timer 创建并返回一个发射周期性事件的定时器。默认情况下，调度动作由 RxJava 线程管理，这意味着定时器线程并非 Vert.x 线程，因此既不在 Vert.x EventLoop 中执行，也不在 Vert.x Worker 线程上执行。

当 RxJava 方法涉及调度器时，通常支持传入一个额外的 io.reactivex.Scheduler。RxHelper.scheduler 方法返回的调度器正可用于此类场景：

Scheduler scheduler = RxHelper.scheduler(vertx);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);

对于阻塞性质的调度操作，可以通过 RxHelper.blockingScheduler 方法创建调度器：

Scheduler scheduler = RxHelper.blockingScheduler(vertx);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);

也可以对 RxJava 进行全局重配置，使其默认使用 Vert.x 调度器：

RxJavaPlugins.setComputationSchedulerHandler(s -> RxHelper.scheduler(vertx));
RxJavaPlugins.setIoSchedulerHandler(s -> RxHelper.blockingScheduler(vertx));
RxJavaPlugins.setNewThreadSchedulerHandler(s -> RxHelper.scheduler(vertx));
RxJava 使用 computation 表示非阻塞任务，使用 io 表示阻塞任务，这与 Vert.x 的术语习惯恰好相反。

Rx 化 的 Vert.x API 同样在 RxHelper 类上提供了类似方法：

Scheduler scheduler = RxHelper.scheduler(vertx);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
RxJavaPlugins.setComputationSchedulerHandler(s -> RxHelper.scheduler(vertx));
RxJavaPlugins.setIoSchedulerHandler(s -> RxHelper.blockingScheduler(vertx));
RxJavaPlugins.setNewThreadSchedulerHandler(s -> RxHelper.scheduler(vertx));

还可以创建基于命名 Worker 线程池的调度器。如果你希望复用特定线程池来调度阻塞动作，这非常实用：

Scheduler scheduler = RxHelper.blockingScheduler(workerExecutor);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);

JSON 反序列化

FlowableHelper.unmarshaller 创建一个 io.reactivex.rxjava3.FlowableOperator，将 JSON 格式的 Flowable<Buffer> 转换为实体对象流：

fileSystem.open("/data.txt", new OpenOptions()).onComplete(result -> {
  AsyncFile file = result.result();
  Flowable<Buffer> observable = FlowableHelper.toFlowable(file);
  observable.compose(FlowableHelper.unmarshaller(MyPojo.class)).subscribe(
      mypojo -> {
        // 处理反序列化后的实体对象
      }
  );
});

通过 Rx 化 辅助工具同样可以实现相同效果：

fileSystem
  .rxOpen("/data.txt", new OpenOptions())
  .flatMapObservable(file -> file.toObservable())
  .compose(ObservableHelper.unmarshaller((MyPojo.class)))
  .subscribe(mypojo -> {
    // 处理反序列化后的实体对象
  });

部署 Verticle

要部署已有的 Verticle 实例，可以使用 RxHelper.deployVerticle，它会部署 Verticle 并返回包含部署 ID 的 Single<String>：

Single<String> deployment = RxHelper.deployVerticle(vertx, verticle);

deployment.subscribe(id -> {
  // 部署成功
}, err -> {
  // 部署失败
});

Rx 化 API（Rxified API）

Rx 化 API 是 Vert.x API 的代码生成版本。该 API 统一使用 io.vertx.rxjava3 前缀，例如原生的 io.vertx.core.Vertx 类被转换为 Vertx 类。

Rx 化 API 以两种风格暴露 Vert.x 异步方法：

  • 原始方法被转换为返回急切求值（Eager）且带缓存订阅的 RxJava 等价方法；

  • 带有 rx 前缀的衍生方法，仅在被订阅时才触发执行原生方法。

// 立即写入
// 无需订阅
// completion 提供异步完成结果
response.write(buffer);

// 此时尚未发生写入
completion = response.rxWrite(buffer);

// 真正执行写入
completion.subscribe(() -> ..., err -> ...);

你可以根据实际需要选择调用原始方法或 Rx 化方法；例如当你无需订阅或者不在意结果时，可以直接调用原始方法。

嵌入式运行 Rx 化 Vert.x

直接调用 Vertx.vertx 方法：

Vertx vertx = io.vertx.rxjava3.core.Vertx.vertx();

作为 Verticle 运行

继承 AbstractVerticle 类，框架会自动完成包装：

class MyVerticle extends AbstractVerticle {
  public void start() {
    // 在此处使用 Rx 化的 Vertx
  }
}

部署 RxJava Verticle 依然由标准的 Java 部署器执行，无需指定专门的部署器。

具备异步启动逻辑的 Verticle 可以改写 rxStart 方法并返回 Completable：

class MyVerticle extends AbstractVerticle {
  public Completable rxStart() {
    return vertx.createHttpServer()
      .requestHandler(req -> req.response().end("Hello World"))
      .rxListen()
      .ignoreElement();
  }
}

常用 API 示例

现在我们来看一些在 Vert.x 中使用 RxJava 的典型示例。

EventBus 消息流

事件总线的 MessageConsumer 天然支持转换为 Observable<Message<T>>：

EventBus eb = vertx.eventBus();
MessageConsumer<String> consumer = eb.<String>consumer("the-address");
Flowable<Message<String>> flowable = consumer.toFlowable();
Disposable sub = flowable.subscribe(msg -> {
  // 接收到消息
});

// 10 秒后注销数据流
vertx.setTimer(10000, id -> {
  sub.dispose();
});

MessageConsumer 提供的是 Message 消息流。 必要时，可以通过 body 直接获取消息正文流：

EventBus eb = vertx.eventBus();
MessageConsumer<String> consumer = eb.<String>consumer("the-address");
Flowable<String> flowable = consumer.bodyStream().toFlowable();

接下来可以使用 RxJava 的 Map/Reduce 组合风格：

Flowable<Double> flowable = vertx.eventBus().
    <Double>consumer("heat-sensor").
    bodyStream().
    toFlowable();

flowable.
    buffer(1, TimeUnit.SECONDS).
    map(samples -> samples.
        stream().
        collect(Collectors.averagingDouble(d -> d))).
    subscribe(heat -> {
      vertx.eventBus().send("news-feed", "Current heat is " + heat);
    });

HTTP 客户端请求

你可以轻松使用 HTTP 客户端发起请求并处理响应：

HttpClient client = vertx.createHttpClient();
client.rxRequest(HttpMethod.GET, 8080, "localhost", "/")
  .flatMap(request -> request
    .rxSend()
    .flatMap(response -> {
      if (response.statusCode() == 200) {
        return response.body();
      } else {
        return Single.error(VertxException.noStackTrace("Invalid response"));
      }
    }))
  .subscribe(body -> {
    // 处理响应正文
  });

当需要处理大型流式响应时，可以从 HTTP 响应中获取 Flowable<Buffer>：

HttpClient client = vertx.createHttpClient();
client.rxRequest(HttpMethod.GET, 8080, "localhost", "/")
  .flatMapPublisher(request -> request
    .rxSend()
    .flatMapPublisher(response -> {
      if (response.statusCode() == 200) {
        return response.toFlowable();
      } else {
        return Flowable.error(VertxException.noStackTrace("Invalid response"));
      }
    }))
  .subscribe(chunk -> {
    // 流式处理响应分块
  });

你也可以使用 Vert.x Web Client。

HTTP 服务端请求

HttpServerRequest 可以适配为 Observable<Buffer>：

Observable<Buffer> observable = request.toObservable();

ObservableHelper.unmarshaller 可用于解析并将 JSON 请求映射为对象：

Flowable<MyPojo> flowable = request.
  toFlowable().
  compose(FlowableHelper.unmarshaller(MyPojo.class));

WebSocket 客户端

connect 在 WebSocket 连接成功时提供单一回调，失败时触发错误回调：

WebSocketClient client = vertx.createWebSocketClient(new WebSocketClientOptions());
client.rxConnect(8080, "localhost", "/the_uri").subscribe(
    ws -> {
      // 使用 websocket
    },
    error -> {
      // 无法连接
    }
);

随后可以轻松将 WebSocket 转换为 Observable<Buffer>：

socketObservable.subscribe(
    socket -> {
      Flowable<Buffer> dataObs = socket.toFlowable();
      dataObs.subscribe(buffer -> {
        System.out.println("Got message " + buffer.toString("UTF-8"));
      });
    }
);

WebSocket 服务端

ServerWebSocket 同样可以轻松转换为 Observable<Buffer>：

socketObservable.subscribe(
    socket -> {
      Flowable<Buffer> dataObs = socket.toFlowable();
      dataObs.subscribe(buffer -> {
        System.out.println("Got message " + buffer.toString("UTF-8"));
      });
    }
);