改造默认线程

Question

我在我的 Android 应用程序中使用 Retrofit 和 RxJava，我的代码：

public void getConfig(NetworkSubscriber subscriber) {
    Observable<Config> observable = mApi.getConfig();
    observable.subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(subscriber);
}

public void getCode(String mobile, int type, NetworkSubscriber subscriber) {
    Observable<BaseMessageEntity> observable = mApi.getCode(mobile, type);
    observable.subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(subscriber);
}

而且我不想写 .subscribeOn(Schedulers.newThread()) 并且 .observeOn(AndroidSchedulers.mainThread()) 每个商业方法

我该怎么办？

Answer 1

您可以使用 compose() 将其缩减为一行。例如，下面是 getConfig() 方法的修改版本。它假定您正在使用 retrolambda。

public void getConfig(NetworkSubscriber subscriber) {
    Observable<Config> observable = mApi.getConfig();
    observable
            .compose(this::setupThreads)
            .subscribe(subscriber);
}

setupThreads() 方法如下所示：

private <T> Observable<T> setupThreads(final Observable<T> observable) {
    return observable
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread());
}

一些参考资料：

RxJava 维基 - Transformational Operators
Dan Lew 博客 post - Don't break the chain: use RxJava's compose() operator

Answer 2

如果您不想在每次调用时都指定您想要的线程，您可以围绕 RxJavaCallAdapterFactory 创建一个包装器以默认设置您的线程。

public class RxThreadCallAdapter extends CallAdapter.Factory {

    RxJavaCallAdapterFactory rxFactory = RxJavaCallAdapterFactory.create();
    private Scheduler subscribeScheduler;
    private Scheduler observerScheduler;

    public RxThreadCallAdapter(Scheduler subscribeScheduler, Scheduler observerScheduler) {
        this.subscribeScheduler = subscribeScheduler;
        this.observerScheduler = observerScheduler;
    }

    @Override
    public CallAdapter<?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
        CallAdapter<Observable<?>> callAdapter = (CallAdapter<Observable<?>>) rxFactory.get(returnType, annotations, retrofit);
        return callAdapter != null ? new ThreadCallAdapter(callAdapter) : null;
    }

    final class ThreadCallAdapter implements CallAdapter<Observable<?>> {
        CallAdapter<Observable<?>> delegateAdapter;

        ThreadCallAdapter(CallAdapter<Observable<?>> delegateAdapter) {
            this.delegateAdapter = delegateAdapter;
        }

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

        @Override
        public <T> Observable<?> adapt(Call<T> call) {
            return delegateAdapter.adapt(call).subscribeOn(subscribeScheduler)
              .observeOn(observerScheduler);
        }
    }
}

然后在您的构建器中使用它代替 RxJavaCallAdapterFactory.create() --

Retrofit retrofit = new Retrofit.Builder()
    .baseUrl("https://api.github.com/")
    .addConverterFactory(GsonConverterFactory.create())
    .addCallAdapterFactory(new RxThreadCallAdapter(Schedulers.io(), AndroidSchedulers.mainThread()))
    .build();

Answer 3

要为 subscribeOn 使用默认调度程序，您可以在创建 Retrofit 实例时将其作为参数直接传递给 RxJavaCallAdapterFactory：

new Retrofit.Builder()
      .client(okHttpClient)
      .addCallAdapterFactory(RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io()))
.build();

（对于 RxJava1 使用 RxJavaCallAdapterFactory）

在 Retrofit 2.0.0 中引入

改造默认线程

Retrofit default thread

android

rx-java

retrofit

rx-android