如何使用 RxJava 2 发出 HTTP 请求来检查内容类型？答案

【问题标题】：How to make a HTTP request to check a content type with RxJava 2?如何使用 RxJava 2 发出 HTTP 请求来检查内容类型？
【发布时间】：2017-07-28 20:23:47
【问题描述】：

我需要从特定 URL 获取内容类型。我知道我们可以通过简单的编码来做到这一点：

URL url = new URL("https://someurl.com");
HttpURLConnection connection = (HttpURLConnection) url.openConnection();
connection.setRequestMethod("HEAD"); // Request Method: GET/POST/UPDATE...
connection.connect();
String contentType = connection.getContentType();

由于这会阻塞 UI 线程（同步操作），如何在 Android 上使用 RxJava 2 发出 HTTP 请求？

注意事项：

我不想使用 AsyncTask 来实现。 Why?
这与 RxJava 2 相关，与版本 1 无关。
如果可以的话，给我一个清晰、简单和简洁的例子。

【问题讨论】：

请注意，Dan Lew 在该博客文章中引用的所有AsyncTask 问题都可以克服，使用 RxJava 2 需要您采取类似的步骤。 RxJava，即使是 RxAndroid，也不是什么“灵丹妙药”。这当然是一种很好的方法，可以随意使用。不过，就个人而言，我不会打扰HttpURLConnection。 FWIW，here is a sample app 使用带有 OkHttp 的 RxJava/RxAndroid 发出 HTTP 请求（特别是来自 GraphQL 端点）。

标签： android rx-java reactive-programming rx-java2

【解决方案1】：

使用 RxJava just 运算符离开主线程并继续计算调度程序的线程处理，然后使用 flatMap 进行 http 调用并查找内容类型，网络调用应在 IO 调度程序的线程上运行，最后在主线程上观察并订阅结果。

Observable.just(1).subscribeOn(Schedulers.computation())
       .flatMap(dummyValueOne -> {
          return Observable.just(getContentType).subscribeOn(Schedulers.io()); 
       }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String contentType) throws Exception {
            //do nextsteps with contentType, you can even update UI here as it runs on main thread
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e("GetContentType", "exception getting contentType", throwable);
                    }
                }));

【讨论】：

【解决方案2】：

您可以使用Callable。这是一个例子，

Observable.fromCallable((Callable<Object>) () -> {
    // do stuff
    return object;
})
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                // so that this can be properly disposed in onDestroy()
                compositeDisposable.add(d);
            }

            @Override
            public void onNext(@NonNull Object object) {
                // do stuff with the result
            }

            @Override
            public void onError(@NonNull Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

【讨论】：