RxJava操作符doOnNext

Posted 2020-07-13 一叶飘舟

doOnNext官方介绍：

The doOnNext operator is much like doOnEach(Action1) except that the Action that you pass it as a parameter does not accept a Notification but instead simply accepts the emitted item.

可以这么理解：

• do系列的作用是side effect,当onNext发生时，它被调用，不改变数据流。
• doOnNext()允许我们在每次输出一个元素之前做一些额外的事情。

我们先通过例子了解下。

有如下场景：

你的程序取到的 User 并不应该直接显示，而是需要先与数据库中的数据进行比对和修正后再显示。

RxJava与Retrofit结合使用，代码如下：

getUser(userId)
    .doOnNext(new Action1<User>() {
        @Override
        public void call(User user) {
            processUser(user);
        })
    .observeOn(androidSchedulers.mainThread())
    .subscribe(new Observer<User>() {
        @Override
        public void onNext(User user) {
            userView.setUser(user);
        }

        @Override
        public void onCompleted() {
        }

        @Override
        public void onError(Throwable error) {
            // Error handling
            ...
        }
    });

后台代码和前台代码全都写在一条链中，清晰很多。在使用User信息之前，在doOnNext方法里对User信息进行了处理，最后执行了subscriber里的OnNext方法。

从github上很多项目和这篇文章来看，do系列的作用

• 使用doOnNext()来调试
• 在flatMap()里使用doOnError()作为错误处理。
• 使用doOnNext()去保存/缓存网络结果

看下面代码：

final SimpleDateFormat sDateFormat = new SimpleDateFormat("yyyy-MM-dd    hh:mm:ss");
        Observable.create(new Observable.OnSubscribe<Person>() {
            @Override
            public void call(Subscriber<? super Person> subscriber) {
                String date = sDateFormat.format(new Date());
                System.out.println(date + " call " + Thread.currentThread().getName());
                Person person = new Person(201);
                subscriber.onNext(person);
            }
        }).subscribeOn(Schedulers.io()) //指定耗时进程
                .observeOn(Schedulers.newThread()) //指定doOnNext执行线程是新线程
                .doOnNext(new Action1<Person>() {
                    @Override
                    public void call(Person person) {
                        String date = sDateFormat.format(new Date());
                        System.out.println(date + " call " + Thread.currentThread().getName());
                        person.age = 301;
                        try {
                            Thread.sleep(2000);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).observeOn(AndroidSchedulers.mainThread())//指定最后观察者在主线程
                .subscribe(new Action1<Person>() {
                    @Override
                    public void call(Person person) {
                        String date = sDateFormat.format(new Date());
                        System.out.println(date + " call " + Thread.currentThread().getName());
                        Log.d(TAG, "call: " + person.age);
                    }
                });

执行结果：

03-01 14:49:29.897 23442-24145/com.example.myrxlearn I/System.out: 2016-03-01    02:49:29 call RxCachedThreadScheduler-2
03-01 14:49:29.907 23442-24144/com.example.myrxlearn I/System.out: 2016-03-01    02:49:29 call RxNewThreadScheduler-2
03-01 14:49:31.907 23442-23442/com.example.myrxlearn I/System.out: 2016-03-01    02:49:31 call main

也就是说直到doOnNext里的方法在新线程执行完毕，subscribe里的call才有机会在主线程执行。

一直没看到有合适的方法解决这个问题，因为缓存的时间不应该去阻碍主线程里数据的显示。

今天做回顾时看到了这篇文章

非阻塞I/O操作

现在我们可以使用Schedulers.io()创建非阻塞的版本：

public static void storeBitmap(Context context, Bitmap bitmap,
                                   String filename) {
        Schedulers.io().createWorker().schedule(() -> {
            blockingStoreBitmap(context, bitmap, filename);
        });
    }

然后想起来一直存在的这个问题。只需要把上面的代码改成

.doOnNext(new Action1<Person>() {
            @Override
            public void call(Person person) {
                String date = sDateFormat.format(new Date());
                System.out.println(date + " call " + Thread.currentThread().getName());
                person.age = 301;
                Schedulers.io().createWorker().schedule(new Action0() {
                    @Override
                    public void call() {
                        try {
                            Thread.sleep(2000);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        })

不需要在用observeOn指定在新线程就可以实现

03-01 14:55:02.307 30368-30406/com.example.myrxlearn I/System.out: 2016-03-01    02:55:02 call RxCachedThreadScheduler-1
03-01 14:55:02.307 30368-30406/com.example.myrxlearn I/System.out: 2016-03-01    02:55:02 call RxCachedThreadScheduler-1
03-01 14:55:02.347 30368-30368/com.example.myrxlearn I/System.out: 2016-03-01    02:55:02 call main

这样doOnNext可以方便的用来调试，用来缓存。

官网说：

doOnNext is for side-effects: you want to react (eg. log) to item
emissions in an intermediate step of your stream, for example before
the stream is filtered, for transverse behavior like logging, but you
still want the value to propagate down the stream.

onNext is more final, it consumes the value.

使用doOnNext的副作用是什么，还需要项目中来发现。

最后补充点。

Observable.create(new Observable.OnSubscribe<Person>() {
            @Override
            public void call(Subscriber<? super Person> subscriber) {
                Person person = new Person(201);
                subscriber.onNext(person);
            }
        }).doOnNext(new Action1<Person>() {
            @Override
            public void call(Person person) {
                person.age = 301;
            }
        }).subscribe(new Action1<Person>() {
            @Override
            public void call(Person person) {
                Log.d(TAG, "call: " + person.age);//输出301
            }
        });

上面的代码执行结果在doOnNext方法执行完，改变了流里的数据。是不是与官方文档说的不一样呢？

参考：http://blog.csdn.net/wangkai0681080/article/details/50772721