RxJava进阶：如何借助其他Observable数据将Callback转为Observable？

问题描述 #

我是RxJava新手，目前遇到一个将Callback转换为Observable的问题。我有一个使用Callback的方法：

client.loadAsync(body, object : Callback {
   override fun onSuccess(response: AwesomeResponse) {
      // Successful response!
   }
   
   override fun onError(exception: Exception) {
      // Error response!
   }
})

我通过Single.create将其转换为Single：

Single.create {
   client.loadAsync(body, object : Callback {
      override fun onSuccess(response: AwesomeResponse) {
         it.onSuccess(response)
      }
   
      override fun onError(exception: Exception) {
         it.onError(exception)
      }
   })
}

但问题在于构建body参数需要另一个Observable的数据。我当前的实现方式如下：

otherObservable.doOnNext { dto -> 
   val body = clientBody(id = dto.Id)

   Single.create {
      client.loadAsync(body, object : Callback {
         override fun onSuccess(response: AwesomeResponse) {
            it.onSuccess(response)
         }

         override fun onError(exception: Exception) {
            it.onError(exception)
         }
      })
   }
   .subscribeOn(scheduler)
   .subscribe(
      { uiReportSubject.onNext(it) }, 
      {}, 
      compositeDisposable
   )
}
.subscribeOn(scheduler)
.subscribe()

我通过订阅uiReportSubject（一个BehaviourSubject）将数据传递到UI层。我想知道这种实现方式是否正确？有没有更优的实现方式，比如使用类似map的操作符？

解决方案 #

你的当前实现不算最优，核心问题是嵌套订阅（在doOnNext内创建并订阅新的Single），这会导致代码可读性差、Disposable管理混乱，也违背了RxJava链式调用的设计初衷。

最优实现：用flatMapSingle串联流 #

RxJava的flatMapSingle操作符专门处理"上游Observable发射数据后，触发另一个Single任务"的场景，完美替代嵌套订阅：

1. 先封装loadAsync为可复用的Single函数（增加Disposed检查避免内存泄漏）：

fun loadData(body: ClientBody): Single<AwesomeResponse> {
    return Single.create { emitter ->
        client.loadAsync(body, object : Callback {
            override fun onSuccess(response: AwesomeResponse) {
                if (!emitter.isDisposed) {
                    emitter.onSuccess(response)
                }
            }

            override fun onError(exception: Exception) {
                if (!emitter.isDisposed) {
                    emitter.onError(exception)
                }
            }
        })
    }.subscribeOn(scheduler)
}

2. 用链式调用串联两个流：

otherObservable
    .flatMapSingle { dto ->
        val body = clientBody(id = dto.Id)
        loadData(body)
    }
    .subscribeOn(scheduler)
    .subscribe(
        { uiReportSubject.onNext(it) },
        { /* 建议补充错误处理，避免异常被静默吞掉 */ },
        compositeDisposable
    )

优化细节说明 #

• 消除嵌套订阅：链式调用让逻辑线性化，可读性和可维护性大幅提升
• 统一资源管理：所有订阅通过compositeDisposable集中管理，避免遗漏导致的内存泄漏
• 增加安全检查：在Callback回调中先判断emitter是否已取消订阅，防止无效回调引发问题
• 规范错误处理：原代码空的错误回调会吞掉异常，优化后建议补充错误处理逻辑，便于问题排查

额外建议：简化数据传递 #

如果UI层仅需要订阅最终的AwesomeResponse，可以直接让UI层订阅上述链式Observable，无需通过BehaviourSubject中转，减少不必要的复杂度。只有当需要多订阅者共享数据、或保留最新状态时，Subject才是必要的选择。

内容的提问来源于stack exchange，提问作者Alfredo Bejarano