1. RxJava 中的异步控制

在以前也遇到了类似的场景,那时候还是再使用AsyncTask的时候。
我们知道AsyncTask中有doInBackground()方法是一个子线程的异步方法。我们一般在里面执行耗时操作。
但是我们会在doInBackground()中执行一个耗时的异步操作吗?看看下面的例子

...
protected boolean doInBackground(String... urls) {
        loadImageFromNetwork(urls[0], targetObej);
        return true;
}
...    

这里的示例中,我们调用loadImageFromNetwork()方法,将第一个参数中的图片下载下来,然后填充到targetObjet这个对象中去。
但是这里调用的实际上是一个异步操作,程序调用了loadImageFromNetwork()就顺序执行了,返回了一个true值。
这样的话再加载情况未知的情况下,程序逻辑已经继续执行下去了。

如果只是使用RxJava来控制业务逻辑的话,那么异步线程里再开一个子线程的问题也会遇到同样问题呀:

/**
 * @author rosu on 2018/10/27
 * 这个类用于展示 Rxjava 中异步操作中继续调用异步方法的情况
 */
public class FlatMapWithChildProcess {
    public static void main(String[] args) {
        Observable
                .create((ObservableOnSubscribe<Integer>) emitter -> {
                    System.out.println("Create1 ===>>> 创建并发射事件");
                    System.out.println("当前线程====>>> 1" + Thread.currentThread().getName());
                    emitter.onNext(1);
                })
                .flatMap((Function<Integer, ObservableSource<Integer>>) integer -> {
                    System.out.println("当前线程====>>> 2" + Thread.currentThread().getName());
                    hardWork();
                    System.out.println("顺序执行了");
                    return Observable.just(integer);
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.out.println("onSubscribe =======>>> 开始订阅事件");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext ======>>> " + integer);

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete ========>>> ");
                    }
                });
    }

    private static void hardWork(){
        new Thread(() -> {
            try {
                System.out.println("当前线程====>>> 3" + Thread.currentThread().getName());
                Thread.sleep(10000);
                System.out.println("睡眠完成的子线程");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

输出为:

onSubscribe =======>>> 开始订阅事件
Create1 ===>>> 创建并发射事件
当前线程====>>> 1main
当前线程====>>> 2main
当前线程====>>> 3Thread-0
顺序执行了
onNext ======>>> 101
睡眠完成的子线程

虽然例子看起来有点长,但内容不多。我们只是尝试在flatMap()调用了hardWork()方法,该方法中中起了一个耗时子线程。
看到输出的结果也在我们的意料之中,原有的工作逻辑在调用了hardWork()之后就继续执行了,因为他无法得知那个方法是个异步的方法,也无法获得该方法的执行状态。

例子比较简单,在实际的工作中我们可能会遇到一些情况,考虑这样一个例子:

  1. 利用 RxJava 循环发射一些事件,常见是用fromArray()intervalRange()这样的方法

    • 此处我们发射一个url链接数组的元素
  2. 我们利用了 RxJava 本身的特性来控制业务逻辑,包括对每个事件的处理

    • 此处,我们可能是对url做一些拼接或者判断有效性的工作
  3. 之后我们需要利用发射的事件做耗时操作

    • 此处,我们是利用url来下载文件,假设调用了download(url:Int)方法
  4. 最后在onComplete()方法中完成视图操作

这看起来是非常正常的业务逻辑,唯一值得注意的地方应该是耗时操作这个地方。我们先来看看例子的简易代码示例:

public class FlatMapWithChildProcess {
    public static void main(String[] args) {
        Observable
                .fromArray(1, 2, 3, 4, 5, 6)
                .flatMap((Function<Integer, ObservableSource<Integer>>) integer -> {
                    download(integer);
                    return Observable.just(integer);
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.out.println("onSubscribe =======>>> 开始订阅事件");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext ======>>> " + integer);

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete ========>>> RxJava 事件完成了");
                        UpdateUI();
                    }
                });
    }

    private static void download(int pos){
        new Thread(() -> {
            try {
                System.out.println("接到工作 ===>>> " + pos);
                Thread.sleep(10000);
                System.out.println("完成工作 ===>>> " + pos + "\n 时间:" + System.currentTimeMillis());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

看一下输出:

onSubscribe =======>>> 开始订阅事件
接到工作 ===>>> 1
onNext ======>>> 1
接到工作 ===>>> 2
onNext ======>>> 2
接到工作 ===>>> 3
onNext ======>>> 3
接到工作 ===>>> 4
onNext ======>>> 4
onNext ======>>> 5
接到工作 ===>>> 5
onNext ======>>> 6
接到工作 ===>>> 6
onComplete ========>>> RxJava 事件完成了
完成工作 ===>>> 1
 时间:1543916254101
完成工作 ===>>> 4
 时间:1543916254103
完成工作 ===>>> 6
 时间:1543916254103
完成工作 ===>>> 3
 时间:1543916254103
完成工作 ===>>> 2
 时间:1543916254103
完成工作 ===>>> 5
 时间:1543916254103

这里可以看到我们耗时操作还没做完,RxJava就已经回调了onComplete()了。所以这显然是不行的。

2. Flowable 能拯救这段代码吗?

众所周知...RxJava2 带来了Flowable这个新的观察者。又一个众所周知,Flowable是一个带有背压控制的观察者。
那么背压控制,能解决这个问题吗?

2.1 Flowable 的背压误区?FlatMap 初探

于是我随手写了这段代码:

public class FlowableWithBackPressure {
    public static void main(String[] args) {
        Flowable
                .fromArray(1, 2, 3, 4, 5)
                .flatMap((Function<Integer, Publisher<Integer>>) integer -> {
                    download(integer);
                    return Flowable.just(integer);
                })
                .map(integer -> integer + 20)
                .subscribe(new FlowableSubscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("RxJava =======>>> onNext");
                    }

                    @Override
                    public void onError(Throwable t) {

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("RxJava =======>>> 时间完成了");
                    }
                });


    }

    private static void download(int pos){
        new Thread(() -> {
            try {
                System.out.println("接到工作 ===>>> " + pos);
                Thread.sleep(500);
                System.out.println("完成工作 ===>>> " + pos + "\n 时间:" + System.currentTimeMillis());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

看一下输出:

接到工作 ===>>> 1
接到工作 ===>>> 2
接到工作 ===>>> 3
接到工作 ===>>> 4
接到工作 ===>>> 5
完成工作 ===>>> 1
 时间:1543918837921
完成工作 ===>>> 3
 时间:1543918837924
完成工作 ===>>> 4
 时间:1543918837924
完成工作 ===>>> 2
 时间:1543918837924
完成工作 ===>>> 5
 时间:1543918837926

咦?貌似有什么地方出了问题啊。按照我的平时认知,没有调用Subscription.request(),就不会发射事件才对啊。
为啥这里还是调用了呢?当然熟悉的人一下子就看出来了。
问题出在与flatmap()方法。

实际上对于flatMap()之类的方法,是将原来的事件流转换为新的类型的事件流。问题就在这里了。
转换的步骤,根据文档的说明是,flatMap()会将每个事件重新包装,最后再将所有事件合并发射。这样的话,实际上就是又构造了一个新的事件发射器,也就是一个新的『上游』

事件的『上游』和『下游』

我之前粗浅的认知里,以为第一个发射的源头是上游,其他都是下游。实际上『上下游』是一个相对的概念,比如这里的flatMap(),他重新包装了事件并重新发射了,他就是一个新的『上游』。这样的话,肯定所有事件不需要request()就可以直接发射到这个flatMap()方法里面了。

2.2 Flowable 和 背压控制

我们了解了『上下游』概念之后,其实横在我们面前的是,如何正确地动态控制事件发射呢?
众所周知,Flowable 带来的背压控制的概念。我们前面也提到了通过Subscription.request()来控制上游发射。
但是类似 2.1 中举的例子,在上游你是无法控制的。而且我们又要利用RxJava来控制业务逻辑,也就是对每个链接进行处理。
这样的话,实际上我们只能在下游动态拉取才行。动态拉取就是Subscription.request()啊?
那该怎么做呢?
其实很简单,我们把耗时操作放在onNext()里就行了:

public class FlowableWithBackPressure {
    private static Subscription mSubscription;

    public static void main(String[] args) {
        Flowable
                .fromArray(1, 2, 3, 4, 5)
                .map(integer -> integer + 20)
                .subscribe(new FlowableSubscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription = s;
                        mSubscription.request(1);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext() =======>>> 调用 download() 方法");
                        download(integer);
                    }

                    @Override
                    public void onError(Throwable t) {

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("RxJava =======>>> 时间完成了");
                    }
                });


    }

    private static void download(int pos){
        new Thread(() -> {
            try {
                System.out.println("接到工作 ===>>> " + pos);
                Thread.sleep(3000);
                System.out.println("完成工作 ===>>> " + pos + "\n 时间:" + System.currentTimeMillis() + "\n 准备拉取");
                mSubscription.request(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

可以看到,任务执行到onNext()方法,然后我们在里面调用了一个异步的耗时操作,等到耗时操作完成之后,采取拉取下一个事件发射。

这是输出结果:

onNext() =======>>> 调用 download() 方法
接到工作 ===>>> 21
完成工作 ===>>> 21
 时间:1543983837259
 准备拉取
onNext() =======>>> 调用 download() 方法
接到工作 ===>>> 22
完成工作 ===>>> 22
 时间:1543983840264
 准备拉取
onNext() =======>>> 调用 download() 方法
接到工作 ===>>> 23
完成工作 ===>>> 23
 时间:1543983843269
 准备拉取
onNext() =======>>> 调用 download() 方法
接到工作 ===>>> 24
完成工作 ===>>> 24
 时间:1543983846275
 准备拉取
onNext() =======>>> 调用 download() 方法
RxJava =======>>> 时间完成了
接到工作 ===>>> 25
完成工作 ===>>> 25
 时间:1543983849278
 准备拉取

这样我们就完成了原来的目标,既用了RxJava控制业务逻辑,又在其中做了耗时操作并动态拉取事件,也就是背压控制。
如果耗时操作并不是业务最下游,那么我们可以使用doOnNext()方法来达到同样的效果:

public class FlowableWithBackPressure {
    private static Subscription mSubscription;

    public static void main(String[] args) {
        Flowable
                .fromArray(1, 2, 3, 4, 5)
                .map(integer -> integer + 20)
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        System.out.println("doOnNext() =======>>> 调用 download() 方法");
                        download(integer);
                    }
                })
                .subscribe(new FlowableSubscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription = s;
                        mSubscription.request(1);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext()2 =======>>>");
                    }

                    @Override
                    public void onError(Throwable t) {

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("RxJava =======>>> 时间完成了");
                    }
                });


    }

    private static void download(int pos){
        new Thread(() -> {
            try {
                System.out.println("接到工作 ===>>> " + pos);
                Thread.sleep(3000);
                System.out.println("完成工作 ===>>> " + pos + "\n 时间:" + System.currentTimeMillis() + "\n 准备拉取");
                mSubscription.request(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

这是输出:

doOnNext() =======>>> 调用 download() 方法
接到工作 ===>>> 21
onNext()2 =======>>>
完成工作 ===>>> 21
 时间:1543984443387
 准备拉取
doOnNext() =======>>> 调用 download() 方法
onNext()2 =======>>>
接到工作 ===>>> 22
完成工作 ===>>> 22
 时间:1543984446390
 准备拉取
doOnNext() =======>>> 调用 download() 方法
onNext()2 =======>>>
接到工作 ===>>> 23
完成工作 ===>>> 23
 时间:1543984449396
 准备拉取
doOnNext() =======>>> 调用 download() 方法
onNext()2 =======>>>
接到工作 ===>>> 24
完成工作 ===>>> 24
 时间:1543984452400
 准备拉取
doOnNext() =======>>> 调用 download() 方法
onNext()2 =======>>>
RxJava =======>>> 事件完成了
接到工作 ===>>> 25
完成工作 ===>>> 25
 时间:1543984455406
 准备拉取

doOnNext()可以注册一个回调,每当ObservableonNext()调用之前就会调用本方法。

3. 还有一点小问题

我们把耗时操作放在了onNext()中调用,也就是调用了download()方法。
但是这样的还是有两个明显的问题:

  1. download()方法和 RxJava 流耦合了,因为用到了Subscription.request()
  2. 最后的一个事件依旧是还未download()完就调用了onComplete(),这是肯定的。因为最后一个事件之后,并不在需要request()了。所以事件流就结束了

    • 看一下上面的输出,先输出了RxJava =======>>> 事件完成了

看到这两个问题,其实我们就该思考这种做法一开始就存在了问题。本身RxJava就良好地支持了异步回调控制的功能,如果非要在异步中再加上异步,造成的问题就是子线程状态难以控制。

其实我们一开始就可以把download()方法写成同步方法,这样的话,利用RxJava本身对线程的控制能力,我们一样可以轻松地实现类似的业务需求。

标签: Android, RxJava, Flowable, 背压, 异步

添加新评论