Android中rxjava的默认调度器

Question

我正在使用Retrofit返回rxjava Observable以进行异步网络调用。

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

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

似乎我总是在IO线程上订阅并在Android主线程上观察。这似乎是所有资源都推崇的最佳实践。也许除了长时间运算之外，我不太明白我们何时会想要偏离这种模式。

有没有一种方法可以通过默认设置subscribeOn和observeOn线程来消除这种样板？

这是rxjava插件的用例吗？（我找不到它们的许多示例。）

我能否通过混淆retrofit执行器来设置网络边界的默认线程？

- dsg

---

请查看我的问题/答案，了解如何使代码更加DRY：https://dev59.com/questions/GoLba4cB1Zd3GeqPlvI2#25576177 - ktusznio

Answer 1

对于Observable响应，Retrofit目前将subscribeOn设置为RestAdapter的HTTP执行程序(提供或默认)。这是为了将RxJava支持添加到现有行为中。

2.0版本的计划是明确设置subscribeOn和observeOn的默认值(无论是两个、一个还是都不设置)。

如果需要链接多个API调用，您不希望总是在主线程上观察，这是您不想要的原因之一。

- Jake Wharton

---

在版本1.9中，我发现在运行仪器测试时，如果没有调用subscribeOn(something)（通常是Schedulers.io()），会导致死锁。这是预期的行为吗？ - njzk2

Answer 2

2015年9月28日发布的Retrofit 2.0.0-beta2版本的更改日志显示，subscribeOn()是在后台运行所必需的。

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

- Xiaozou

Answer 3

是的，可以同时移除这两个调用。

这是一个Retrofit适配器类，它自动安排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时使用这个适配器：

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

我写了这篇博客文章，详细介绍了这里到底发生了什么。

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

- Greg Ennis

Answer 4

这并不是您寻找的完整答案，但至少可以减轻subscribeOn(Schedulers.io())带来的负担。

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

- Ramdane Oualitsen