RxAndroid技术问询：重复调用方法时如何向Observer发射新项？

Hey Lisa, let's fix this issue so your search queries always deliver fresh results without old subscriptions messing things up! The core problem with your current code is that every time onQueryTextChange runs, you create a new subscription but never cancel the previous ones. This can lead to outdated search results popping up after newer ones, plus potential memory leaks. Here are two solid approaches to resolve this:

1. 手动管理订阅（简单直接） #

We'll use a CompositeDisposable to track all active subscriptions, and clear them before starting a new search to avoid stale results.

// 在你的类中声明一个CompositeDisposable来统一管理订阅
private val compositeDisposable = CompositeDisposable()

fun onQueryTextChange(newText: String?): Boolean {
    // 先取消之前的所有订阅，避免旧请求的结果干扰新查询
    compositeDisposable.clear()
    
    // 处理空文本场景（比如用户清空输入，你可以在这里添加清空UI结果的逻辑）
    newText?.takeIf { it.isNotBlank() }?.let { query ->
        val disposable = model
            .search(query)
            .subscribeOn(workerThreadScheduler)
            .observeOn(mainThreadScheduler)
            .subscribe(observer)
        
        // 将新订阅加入管理容器
        compositeDisposable.add(disposable)
    }
    return true
}

// 记得在组件生命周期结束时清理所有订阅，比如Activity的onDestroy方法
override fun onDestroy() {
    super.onDestroy()
    compositeDisposable.dispose()
}

2. 用RxJava流自动切换订阅（更优雅的Rx风格） #

For a more idiomatic RxJava solution, we'll use a PublishSubject to turn query text changes into an observable stream, then leverage switchMap to automatically cancel previous search subscriptions when a new query comes in. This also lets you add handy optimizations like request debouncing.

// 声明一个Subject来发射查询文本事件
private val querySubject = PublishSubject.create<String>()
private val compositeDisposable = CompositeDisposable()

// 在初始化阶段配置好整个流的处理逻辑
init {
    querySubject
        // 防抖：等待300ms再发射请求，避免用户输入过程中频繁触发搜索
        .debounce(300, TimeUnit.MILLISECONDS)
        // 避免重复的相同查询触发无效请求
        .distinctUntilChanged()
        // 自动切换到新的搜索Observable，同时取消前一个的订阅
        .switchMap { query ->
            model.search(query)
                .subscribeOn(workerThreadScheduler)
        }
        .observeOn(mainThreadScheduler)
        .subscribe(observer)
        .let { compositeDisposable.add(it) }
}

fun onQueryTextChange(newText: String?): Boolean {
    // 只需要把新的查询文本发射到Subject中即可
    newText?.let { querySubject.onNext(it) }
    return true
}

// 生命周期结束时清理所有资源
override fun onDestroy() {
    super.onDestroy()
    compositeDisposable.dispose()
    querySubject.onComplete()
}

关键细节说明 #

• switchMap是核心：它会在新的Observable发出时，立即取消前一个Observable的订阅，确保只有最新的搜索请求在运行，完美解决旧结果干扰的问题。
• 防抖与去重：debounce和distinctUntilChanged是可选但实用的优化，能减少不必要的网络请求或数据库查询，提升整体性能。
• 生命周期管理：无论哪种方案，都要在组件销毁时调用dispose()，防止内存泄漏。

内容的提问来源于stack exchange，提问作者Lisa Anne