RxJava 在订阅中处理 Observable答案

【问题标题】：RxJava dispose Observable in subscribeRxJava 在订阅中处理 Observable
【发布时间】：2020-03-04 14:36:04
【问题描述】：

是否有任何清晰而聪明的解决方案可以从其订阅方法中处理 observable？例如。我有一个 Observable.interval 每秒发出 Long 。在我的订阅方法中，我想检查是否比取消订阅还有 20 秒。

val observable = Observable.interval(1000,TimeUnit.MILLISECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                { num ->

                    if(num == 20.toLong())
                    {
                        observable.dispose()//I cant call dispose here because variable 'observable' is not avaliable yet
                    }
                })

实现这个逻辑的最简单和正确的方法是什么？

【问题讨论】：

您尝试过takeUntil 或takeWhile 运算符吗？ reactivex.io/documentation/operators/takeuntil.html
Gralls 谢谢，是的 takeUntil 正是我想要的。

标签： android kotlin rx-java

【解决方案1】：

我找到了一个简单的解决方案。只需将变量声明和初始化分为两步。

例如

var observable:Disposable? = null
observable = Observable.interval(1000,TimeUnit.MILLISECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                { num ->

                    if(num == 20.toLong())
                    {
                        observable?.dispose()
                    }
                })

【讨论】：