注册

去登录

扫码关注公众号
发送“我爱安卓”
即可获取验证码

解锁回答区域

扫码关注公众号
发送“我爱安卓”

若你登陆，将永久解锁；
若未登录，仅本机解锁。

解锁回答区域

获取注册验证码

扫码关注公众号
发送“我爱安卓”
即可获取验证码

每日一问如何检测和避免 RxJava 的重复线程切换？

xiaoyang 2021-01-31 16:58 收藏

本题目摘自一篇 Blog，大家先尝试思考下，过两天注明博客链接。

众所周知，对于 RxJava我们可以使用：

observeOn
subscribeOn

去做线程调度，我们看个例子：

fun fetchItem(): Single<Item> {
    return download(itemId.getAndIncrement())
        .flatMap(::unZip)
        .flatMap(::checkMd5)
}

private fun download(id: Int): Single<Item> {
    return Single.just(id)
        //Simulate a long time operation
        .delay(300, TimeUnit.MILLISECONDS, Schedulers.io())
        .map { Item(it) }
}

private fun unZip(item: Item): Single<Item> {
    return Single.just(item)
        //Simulate a long time operation
        .delay(300, TimeUnit.MILLISECONDS, Schedulers.io())
}

private fun checkMd5(item: Item): Single<Item> {
    return Single.just(item)
        //Simulate a long time operation
        .delay(300, TimeUnit.MILLISECONDS, Schedulers.io())
}

上面模拟的操作中，download unZip checkMd5 都各自指定了调度器，导致 fetchItem 实际上发生了三次线程切换。

对于这种一系列的耗时操作来说，完全可以运行在同一条后台线程上。

还有这个更刻意的例子：

Observable
    .create(emitter -> {
        System.out.println("create on " + Thread.currentThread().getName());
        emitter.onNext("Test");
        emitter.onComplete();
    })
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.io())
    .map(s -> {
        System.out.println("map on " + Thread.currentThread().getName());
        return s;
    })
    .observeOn(Schedulers.io())
    .flatMapCompletable(s -> {
        System.out.println("flatMap on " + Thread.currentThread().getName());
        return Completable.complete();
    })
    .subscribe();

其实几次操作都交给了 io 线程调度，大概率运行在不同的线程上，实际上完全可以避免没有必要的 io 调度。

问题来了：

我们如何无感知的避免这类多余的线程调度，例如当前方法已经运行在 io 线程，那么就没有必要再去做 io 线程调度了；
无感知指的是，不需要刻意去修改业务逻辑代码，毕竟上面第一个例子那个几个方法内部指定Schedulers.io()也是合理的。

上一问下一问有新回答请提醒我请先登录...

gogogo 2021-01-21 14:01

减少RxJava中多余的线程调度

😁

wangzhengyi : @gogogo

牛皮的搜索能力，点赞

2021-01-27 20:17 回复

Janking 2021-01-18 01:23

刚学完RxJavaPlugins，代码没有验证过，感觉应该差不多就是这么干

    /**
     * 检测重复线程调度器
     */
    private void hook(){
        RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable<?>>() {
            @Override
            public Observable<?> apply(@NonNull Observable observable) throws Exception {
                //获取到正在操作的observeOn
                if (observable instanceof ObservableObserveOn){
                    //获取到正在操作的scheduler
                    Scheduler scheduler = getScheduler((ObservableObserveOn<?>)observable);
                    //获取到上一步的observeOn
                    ObservableSource<?> source = ((ObservableObserveOn<?>)observable).source();
                    if (source instanceof ObservableObserveOn){
                        //获取到上一步的scheduler
                        Scheduler sourceScheduler = getScheduler((ObservableObserveOn<?>)source);
                        if (scheduler.getClass().equals(sourceScheduler.getClass())){
                            //这俩用的是同一个线程调度器，直接返回上一步的Observable包装,本次不需要再包装一层了
                            return (Observable<?>)source;
                        }
                    }
                }
                return observable;
            }
        });
    }

    private Scheduler getScheduler(ObservableObserveOn<?> observableObserveOn) throws Exception{
        Field schedulerField = ObservableObserveOn.class.getField("scheduler");
        schedulerField.setAccessible(true);
        return (Scheduler)schedulerField.get(observableObserveOn);
    }

登录

注册

解锁回答区域

获取注册验证码

每日一问如何检测和避免 RxJava 的重复线程切换？

更多问答 >>

联系我们

目录

删除留言

登录

注册

解锁回答区域

获取注册验证码

每日一问 如何检测和避免 RxJava 的重复线程切换？

更多问答 >>

联系我们

目录

删除留言

每日一问如何检测和避免 RxJava 的重复线程切换？