在RxJava中调用dispose（）之后接收onNext（）中的项

Question

这是我读到的关于dispose() here的内容：

简而言之，当Disposable（由TestObserver实现）被释放时，Observer（也是TestObserver）将不再接收来自Observable的值。

这是我的代码：

private fun createObservableWithDisposable() {
    Observable
            .create { e: ObservableEmitter<String> ->
                val worker = Schedulers.io().createWorker()
                e.setDisposable(worker)
                worker.schedule {
                    for (i in 1..5) {
                        if (i == 3) {
                            worker.dispose()
                            // https://medium.com/@vanniktech/rxjava-2-disposable-under-the-hood-f842d2373e64
                            // After calling dispose(), the subscriber no longer receives items passed in OnNext().
                            // But it doesn't work in my code
                        }
                        e.onNext("Event $i on thread ${Thread.currentThread().name}")
                    }
                }
            }
            .subscribe(
                    { s ->
                        Log.d(TAG, "createObservableWithDisposable onNext msg=$s")
                    },
                    { e ->
                        Log.d(TAG, "createObservableWithDisposable", e)
                    },
                    {
                        Log.d(TAG, "createObservableWithDisposable onComplete")
                    }
            )
}

这就是我在Logcat中看到的：

2019-02-25 08:10:53.414 12071-12101/ru.sample D/RxJavaSamples: createObservableWithDisposable onNext msg=Event 1 on thread RxCachedThreadScheduler-1
2019-02-25 08:10:53.414 12071-12101/ru.sample D/RxJavaSamples: createObservableWithDisposable onNext msg=Event 2 on thread RxCachedThreadScheduler-1
2019-02-25 08:10:53.414 12071-12101/ru.sample D/RxJavaSamples: createObservableWithDisposable onNext msg=Event 3 on thread RxCachedThreadScheduler-1
2019-02-25 08:10:53.414 12071-12101/ru.sample D/RxJavaSamples: createObservableWithDisposable onNext msg=Event 4 on thread RxCachedThreadScheduler-1
2019-02-25 08:10:53.414 12071-12101/ru.sample D/RxJavaSamples: createObservableWithDisposable onNext msg=Event 5 on thread RxCachedThreadScheduler-1

我预计只会看到前两种排放。即，我认为在调用dispose()之后，onNext()将不会被调用。

Answer 1

你处置了提供物品的工人，而不是订户。

要停止接收项目，请尝试

val compositeDisposable = CompositeDisposable()
    compositeDisposable.add(
            Observable
                    .create { e: ObservableEmitter<String> ->
                        val worker = Schedulers.io().createWorker()
                        e.setDisposable(worker)
                        worker.schedule {
                            for (i in 1..5) {
                                if (i == 3) {
                                    compositeDisposable.dispose() //changed here
                                }
                                e.onNext("Event $i on thread ${Thread.currentThread().name}")
                            }
                        }

                    }
                    .subscribe(
                            { s ->
                                Log.d(TAG, "createObservableWithDisposable onNext msg=$s")
                            },
                            { e ->
                                Log.d(TAG, "createObservableWithDisposable", e)
                            },
                            {
                                Log.d(TAG, "createObservableWithDisposable onComplete")
                            }

                    )
    )