改造默认线程

Question

我用Retrofit与RxJava我的Android应用程序,和我的代码:

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

如果您不想在每次调用时指定所需的线程,则可以创建一个包装器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 2

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

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

在Retrofit 2.0.0中介绍过

Answer 3

您可以使用它将其减少到一行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 Wiki - 转换运算符
• Dan Lew博客文章 - 不要破坏链:使用RxJava的compose()运算符