java - 使用 ReactiveX for Java 进行 Http 调用

Question

我是 ReactiveX for Java 的新手，我有以下代码块可以进行外部 http 调用，但它不是异步的。我们正在使用 rxjava 1.2 和 Java 1.8

  private ResponseEntity<String> callExternalUrl(String url, String json, HttpMethod method) {

    RestTemplate restTemplate;
    HttpEntity request;

      request = new HttpEntity(jsonContent, httpHeaders);

    return restTemplate.exchange(url, httpMethod, request, String.class);

  }

我在网上找到了以下代码块，但我无法完全理解它以及如何将它应用到我的代码库中。

private RxClient<RxObservableInvoker> httpClient;
public <T> Observable<T> fetchResult(String url, Func1<Response, T> mapper) {

    return httpClient.target(url)
        .request()
        .rx()
        .get()
        .subscribeOn(Schedulers.io())
        .map(mapper);
  }

Answer 1

如果我没理解错的话，你需要这样的东西来包装你现有的 callExternalUrl

static Observable<String> callExternalUrlAsync(String url, String json, HttpMethod method)
{
    return Observable.fromCallable(() -> callExternalUrl(url, json, method))
            .subscribeOn(Schedulers.io())
            .flatMap(re -> {
                         if (re.hasBody())
                             return Observable.just(re.getBody());
                         else
                             return Observable.error(new RuntimeException("Bad response status " + re.getStatusCode()));
                     },
                     e -> Observable.error(e),
                     (Func0<Observable<? extends String>>) (() -> Observable.empty())) // I need explicit cast or it won't compile :-(
            .observeOn(Schedulers.computation());
}

代码的简短描述:

1. 它安排执行现有的 callExternalUrl在 Schedulers.io 上
2. 对 ResponseEntity<T> 进行最小转换进入成功T和错误案例。它发生在 io 上调度程序也是如此，但这并不重要，因为它真的很短。 (如果 callExternalUrl 中有异常，则按原样传递。)
3. 订阅要在 Schedulers.computation 上执行的结果

注意事项:

1. 您可能想为这两个 subscribeOn 使用自定义调度程序和 observeOn
2. 您可能希望在传递给 flatMap 的第一个 lambda 中有一些更好的逻辑区分成功和错误，您肯定需要一些更具体的异常类型。

高阶魔法

如果您愿意使用高阶函数并牺牲一点性能来减少代码重复，您可以这样做:

// Universal wrapper method
static <T> Observable<T> wrapCallExternalAsAsync(Func3<String, String, HttpMethod, ResponseEntity<T>> externalCall, String url, String json, HttpMethod method)
{
    return Observable.fromCallable(() -> externalCall.call(url, json, method))
            .subscribeOn(Schedulers.io())
            .flatMap(re -> {
                         if (re.hasBody())
                             return Observable.just(re.getBody());
                         else
                             return Observable.error(new RuntimeException("Bad response status " + re.getStatusCode()));
                     },
                     e -> Observable.error(e),
                     (Func0<Observable<? extends T>>) (() -> Observable.empty())) // I need explicit cast or it won't compile :-(
            .observeOn(Schedulers.computation());
}

static Observable<String> callExternalUrlAsync_HigherOrder(String url, String json, HttpMethod method)
{
    return wrapCallExternalAsAsync(MyClass::callExternalUrl, url, json, method);
}

在哪里MyClass你的callExternalUrl在哪里是。

---

更新(仅限异步调用)

私有(private)静态 RxClient httpClient = Rx.newClient(RxObservableInvoker.class);//在这里你可以传递自定义的 ExecutorService

private <T> Observable<String> executeHttpAsync(String url, String httpMethod, Entity<T> entity) {
    return httpClient.target(url)
            .request()
            .headers(httpHeaders) // assuming httpHeaders is something global as in your example
            .rx()
            .method(httpMethod, entity)
            .map(resp -> {
                if (200 != resp.getStatus()) {
                    throw new RuntimeException("Bad status code " + resp.getStatus());
                } else {
                    if (!resp.hasEntity()) {
                        // return null; // or error?
                        throw new RuntimeException("Empty response"); // or empty?
                    } else {
                        try {
                            return resp.readEntity(String.class);
                        } catch (Exception ex) {
                            throw new RuntimeException(ex); // wrap exception into unchecked
                        }
                    }
                }
            })
            .observeOn(Schedulers.computation());
}

private Observable<String> executeGetAsync(String url) {
    return executeHttpAsync(url, "GET", null);
}

private Observable<String> executePostAsync(String url, String json) {
    return executeHttpAsync(url, "POST", Entity.json(json));
}

类似的注意事项同样适用:

1. 您可能想为这两个 newClient 使用自定义调度程序调用并observeOn
2. 您可能希望有一些更好的错误处理逻辑，而不仅仅是检查它是否为 HTTP 200，并且您肯定需要一些更具体的异常类型。但这都是特定于业务逻辑的，因此由您决定。

从你的例子中也不清楚请求的主体( HttpEntity )是如何构建的，以及你是否真的总是想要 String作为您原始示例中的响应。不过我只是照原样复制了你的逻辑。如果您需要更多信息，您可能应该引用 https://jersey.java.net/documentation/2.25/media.html#json 上的文档。