RxJava - 如何在新项目发出时停止正在运行的任务答案

【问题标题】：RxJava - How to stop a running task when new item emittedRxJava - 如何在新项目发出时停止正在运行的任务
【发布时间】：2020-10-29 08:58:58
【问题描述】：

我正在使用 RxJava 为我的 Android 应用程序构建搜索功能。当用户在EditText 中键入/更改查询字符时，发射器将发出新的查询文本，我在我的数据库中搜索以获取与查询匹配的结果。

getEditTextObservable()
                .subscribeOn(Schedulers.computation())
                .debounce(500, TimeUnit.MILLISECONDS)
                .filter {
                    !it.isNullOrEmpty()
                }
                .map {
                    //start searching
                    getResultsInDatabase(it) //this function takes a long time to complete
                }.observeOn(AndroidSchedulers.mainThread())
                .subscribe{
                    //render results on screen
                }

方法getResultsInDatabase(string:String) 需要很长时间才能完成，当查询文本更改时，我想停止方法getResultsInDatabase(string:String)（以防它与先前发出的查询一起运行）以使用新查询再次运行它。我该怎么做才能做到这一点？我会很感激你的帮助。感谢您阅读我的问题。

【问题讨论】：

标签： android kotlin rx-java

【解决方案1】：

您需要switchMap。

getEditTextObservable()
    .subscribeOn(Schedulers.computation())
    .debounce(500, TimeUnit.MILLISECONDS)
    .filter { !it.isNullOrEmpty() } //Be careful with this. Should clearing text clear results too?
    .switchMap { 
        Observable.fromCallable { getResultsInDatabase(it) }.subscribeOn(Schedulers.io()) 
    }.observeOn(AndroidSchedulers.mainThread())
    .subscribe{
        //render results on screen
    }

【讨论】：

谢谢你，我会尽力给你反馈的。
我按照您的建议将map 更改为switchMap，但它没有按我的预期工作。
这或多或少是我用 Rx 编写自己的搜索函数的方式，所以这很令人惊讶。 switchMap 通过在上游发生变化时创建新订阅来工作。这意味着，当查询更改时出现新的发射时，它会放弃先前的订阅以获取您的数据库结果，并启动一个新链。也许您应该添加一个使用线程池 (io()) 的调度程序来获取您的结果，并将其添加到 fromCallable{ }。
谢谢你，我关注你的评论，现在效果很好。只需修改您的答案，我会接受。顺便说一句，你能详细解释一下吗？
我玩过switchMap操作员。我意识到subscribeOn(Schedulers.io()) 只是防止fromCallable 块中的代码在新项目发出之前执行的语句。即使我使用 subscribeOn(AndroidSchedulers.mainThread()) 并删除 subscribeOn、debounce 运算符，您的答案仍然有效。

【解决方案2】：

您可以使用valve 在RxJava 中暂停流。根据它的值，流继续发射或将保持在暂停状态。

【讨论】：

谢谢。我是 RxJava 新手，我会搜索valve 并尝试应用它。