Android RxJava实现首个API响应后循环批量请求并执行后续操作

• 嵌套订阅：违反RxJava最佳实践，在subscribe回调中再次发起订阅会导致事件流断裂，无法统一管理生命周期、异常捕获也会失效，还容易引发内存泄漏。
• 流转换错误：使用map将equipno列表转换为List<Observable<PredictionModel>>后，没有将该列表转换为统一的Observable事件流，直接在回调中处理逻辑脱离了RxJava的流管理。
• 空安全缺失：原代码没有处理equipno为null的边界情况，可能导致后续接口调用参数非法。
• 线程调度不合理：原代码在主线程中发起多个接口请求的订阅，会造成不必要的线程切换开销。

val predictiveMaintenanceServiceCall = PredictiveMaintenanceServiceCall(context)
val currentDate = dateFormatter.format(Calendar.getInstance().time)

predictiveMaintenanceServiceCall.getModels(this)
    // 第一步：提取非空的equipno列表
    .map { response ->
        response.predictionMessage?.models
            ?.mapNotNull { model -> model.equipno } // 过滤掉null的equipno
            ?: emptyList()
    }
    // 第二步：将equipno列表转换为所有getPrediction请求的Observable
    .flatMap { equipnoList ->
        // 遍历生成每个equipno对应的请求Observable
        val requestList = equipnoList.map { equipno ->
            predictiveMaintenanceServiceCall.getPrediction(equipno, currentDate)
                .subscribeOn(Schedulers.io()) // 每个请求都在IO线程执行
        }
        // 合并所有请求，等所有请求完成后发射结果列表
        Observable.zip(requestList) { args -> args.toList() as List<PredictionModel> }
    }
    // 统一线程调度
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    // 统一订阅处理结果
    .subscribe(
        { predictionList ->
            // 所有请求完成，在这里处理所有结果
            Log.i("v ", "loadAlertsList 总请求数：${predictionList.size}")
            predictionList.forEach { prediction ->
                Log.i("v ", "loadAlertsList equipno：${prediction.message?.equipno}")
            }
            // 在这里添加所有请求完成后的自定义操作
        },
        { error ->
            // 统一处理所有环节的异常
            Log.e("v ", "loadAlertsList 出错：${error.message}", error)
        }
    )

• 上述方案默认请求并发执行，所有getPrediction请求会并行发起，全部返回后才会发射结果列表，符合多数场景的性能需求。
• 如果需要按顺序串行发起请求，可以将zip部分替换为Observable.fromIterable(equipnoList).concatMap { predictiveMaintenanceServiceCall.getPrediction(it, currentDate).subscribeOn(Schedulers.io()) }.toList().toObservable()即可。
• 记得在页面销毁时调用Disposable的dispose()方法切断订阅，避免内存泄漏。

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