java - RX Java Single 未从 Single.merge 返回

Question

我有几个 api 调用(Rx singles)，我想将它们组合成一个 Single。我正在使用 Single.merge 尝试合并这些调用的结果，但是当我订阅响应时，我得到一个空数组，因为订阅已经发生。我调用 HealthChecker 期望订阅将返回结果列表:

     new HealthChecker(vertx)
        .getHealthChecks(endpoints)
        .subscribe(messages -> {
            log.info("Completed health check {}", messages);
            routingContext.response()
                          .putHeader("content-type", "text/json")
                          .end(messages.toString());
        });

健康检查器类执行逻辑:

public class HealthChecker {

    private static Logger log = LoggerFactory.getLogger(HealthChecker.class);

    private Vertx vertx;
    private WebClient client;

    public HealthChecker(Vertx vertx) {
        this.vertx = vertx;
        client = WebClient.create(vertx);
    }

    public Single<List<String>> getHealthChecks(JsonArray endpoints) {
        return Single.fromCallable(() -> {

            List<Single<String>> healthChecks = endpoints
                .stream()
                .map(endpoint -> getHealthStatus(client, endpoint.toString()))
                .collect(Collectors.toList());

            return consumeHealthChecks(healthChecks).blockingGet();

        });
    }

    private Single<List<String>> consumeHealthChecks(List<Single<String>> healthChecks) {
        return Single.fromCallable(() -> {
            List<String> messages = new ArrayList<>();

            Single.merge(healthChecks)
                  .timeout(1500, TimeUnit.MILLISECONDS)
                  .subscribe(message -> {
                      log.info("Got health check {}", message);
                      messages.add(message);
                  }, error -> {
                      log.info("Timeout - could not get health check");

                  });

            return messages;
        });
    }

    private Single<String> getHealthStatus(WebClient client, String endpoint) {
        log.info("getting endpoint {}", endpoint);

        return client
            .getAbs(endpoint)
            .rxSend()
            .map(HttpResponse::bodyAsString)
            .map(response -> response);

    }
}

我希望返回值是一个列表，除了我得到的只是一个空列表，然后是结果。这是日志:

09:12:06.235 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - getting endpoint http://localhost:5000/status
09:12:06.241 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - getting endpoint http://localhost:5001/status
09:12:06.241 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - getting endpoint http://localhost:5002/status
09:12:06.241 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - getting endpoint http://localhost:5003/status
09:12:06.241 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - getting endpoint http://localhost:5004/status
09:12:06.300 [vert.x-eventloop-thread-1] INFO  sys.health.HealthCheckVerticle - Completed health check []
09:12:06.688 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - Got health check {"isHealthy":true}
09:12:06.844 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - Got health check {"isHealthy":true}
09:12:06.898 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - Got health check {"isHealthy":false}
09:12:07.072 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - Got health check {"isHealthy":true}
09:12:07.255 [vert.x-eventloop-thread-1] INFO  sys.health.HealthChecker - Got health check {"isHealthy":true}

Answer 1

为什么使用fromCallable和blockingGet？此外，您还可以触发合并，而无需实际等待它运行完成，因此列表为空。相反，在内部 Single 上进行组合:

public Single<List<String>> getHealthChecks(JsonArray endpoints) {
    return Single.defer(() -> {

        List<Single<String>> healthChecks = endpoints
            .stream()
            .map(endpoint -> getHealthStatus(client, endpoint.toString()))
            .collect(Collectors.toList());

        return consumeHealthChecks(healthChecks);
    });
}

private Single<List<String>> consumeHealthChecks(List<Single<String>> healthChecks) {
    return Single.merge(healthChecks)
                 .timeout(1500, TimeUnit.MILLISECONDS)
                 .toList();
}