Android 上 rxjava 的默认调度程序答案

【问题标题】：Default Schedulers for rxjava on AndroidAndroid 上 rxjava 的默认调度程序
【发布时间】：2019-02-09 08:29:55
【问题描述】：

我正在使用 Retrofit 为我的异步网络调用返回 rxjava Observable。

我发现自己在重复以下调用：

someApiCall().subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())

似乎我总是订阅 IO 线程并观察 Android 主线程。这似乎是我发现的所有资源都提倡的最佳实践。也许除了长时间运行的计算之外，我不太明白我们什么时候想要偏离这种模式。

有没有办法通过默认 subscribeOn 和 observeOn 线程来删除这个样板？

这是rxjava plugins 的用例吗？（我找不到很多使用它们的例子。）

我可以通过弄乱retrofit executors来在网络边界设置默认线程吗？

【问题讨论】：

在此处查看我的问题/答案，了解如何稍微干燥一下：stackoverflow.com/a/25576177

标签： java android multithreading retrofit rx-java

【解决方案1】：

对于Observable 响应，Retrofit 当前将subscribeOn 设置为RestAdapter 的HTTP 执行器（提供或默认）。这样做是为了将 RxJava 支持填充到现有行为中。

2.0 的计划是提供为 subscribeOn 和 observeOn 显式设置默认值的能力（无论是两者、只有一个还是两者都没有）。

您不希望总是在主线程上进行观察的一个原因是，例如，如果您需要将多个 API 调用链接在一起。

【讨论】：

在 1.9 版中，我发现在仪器测试死锁期间不调用 subscribeOn(something)（通常是 Schedulers.io()）。这是预期的行为吗？

【解决方案2】：

Retrofit Version 2.0.0-beta2 (2015-09-28) 的 Change Log 显示 subscribeOn() 是在后台运行所必需的。

修复：请求的 Observable 和 Single-based 执行现在行为同步（因此需要 subscribeOn() 才能在后台运行）。

【讨论】：

【解决方案3】：

是的，可以删除这两个调用。

以下是改装适配器类，它自动安排 subscribeOn 和 observedOn 以消除每次调用中对样板调用的需要：

public class RxThreadingCallAdapterFactory extends CallAdapter.Factory {
    private final RxJava2CallAdapterFactory original;

    private RxThreadingCallAdapterFactory() {
        // Always call on background thread
        original = RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io());
    }

    public static CallAdapter.Factory create() {
        return new RxThreadingCallAdapterFactory();
    }

    @Override
    public CallAdapter<?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
        return new RxCallAdapterWrapper(original.get(returnType, annotations, retrofit));
    }

    private static class RxCallAdapterWrapper implements CallAdapter<Observable<?>> {
        private final CallAdapter<?> wrapped;

        public RxCallAdapterWrapper(CallAdapter<?> wrapped) {
            this.wrapped = wrapped;
        }

        @Override
        public Type responseType() {
            return wrapped.responseType();
        }

        @Override
        public <R> Observable<?> adapt(Call<R> call) {
            Observable observable = (Observable) wrapped.adapt(call);

            // Always handle result on main thread
            return observable.observeOn(AndroidSchedulers.mainThread());
        }
    }
}

然后在配置改造时使用这个适配器：

Retrofit.Builder()
    .baseUrl(...)
    .addCallAdapterFactory(RxThreadingCallAdapterFactory.create())

我写了this blog post，其中详细介绍了这里发生的事情。

这将删除我认为是样板的两个调用。我认为 Jake 将后台调用链接在一起的场景并不真正适用，因为在这种情况下，我会改进同步调用并且根本不使用调度程序。

【讨论】：

【解决方案4】：

这不是您正在寻找的完整答案，但这至少消除了纠正subscribeOn(Schedulers.io())的负担

retrofit = new Retrofit
            .Builder()
            .baseUrl(app.getUrlBase())
            .client(httpClient)
            .addCallAdapterFactory(
RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io()) // <-- default subscribeOn() 
)
                .addConverterFactory(jsonFactory)
                .build();

【讨论】：