Android Rx：如何非阻塞处理多API请求队列？

核心思路 #

要实现「单个请求失败不阻塞队列」+「失败请求缓存重发」，核心是让每个请求的错误被独立隔离处理，不让单个请求的异常扩散到整个队列流中。同时需要一套可靠的缓存机制存储失败请求的完整上下文，后续按需触发重发。

比递归更优的实现方案 #

递归方案容易陷入嵌套逻辑，且难以控制并发和队列状态，推荐以下几种Rx原生操作符组合的方案：

1. 用 concatMapEager 处理有序请求队列

如果需要保证请求的执行顺序，concatMapEager 是比普通 concatMap 更高效的选择：它会同时订阅所有请求Observable，但严格按原始顺序发射结果。每个请求的错误可以在内部捕获，不会中断整个队列。

• 每个请求的错误逻辑单独处理：失败时存入缓存，成功时移除缓存项
• 用 onErrorResumeNext(Observable.empty()) 让单个请求失败后，队列继续处理下一个请求

2. 结合 BehaviorSubject 实现可动态触发的任务流

用 BehaviorSubject 作为队列的「触发器」，不管是首次发起请求，还是后续重发缓存的请求，都可以通过向Subject发送信号来触发执行。这种方式可以灵活控制队列的启停、新增任务：

• 当有新请求需要执行时，将请求对象发送到Subject
• 订阅Subject的流，用 flatMap 或 concatMapEager 处理每个请求
• 失败请求存入缓存，待网络恢复或手动触发时，将缓存的请求重新发送到Subject

3. 用 Completable.mergeDelayError 批量并行重发

如果不需要严格的执行顺序，批量重发缓存请求时可以用 mergeDelayError：它会并行执行所有请求，某个请求失败不会影响其他请求的执行，最后统一收集所有错误。这种方式效率更高，适合对顺序无要求的场景。

代码示例（Kotlin） #

// 定义失败请求的缓存接口
interface FailedRequestCache {
    fun save(request: FailedPostRequest)
    fun getAll(): List<FailedPostRequest>
    fun remove(request: FailedPostRequest)
}

// 失败请求的模型，保存重发所需的全部信息
data class FailedPostRequest(
    val url: String,
    val body: RequestBody,
    val headers: Headers,
    val tag: String? = null // 可选：标记请求类型，方便分类处理
)

// 单个POST请求的执行逻辑：失败缓存，成功移除缓存
private fun executePost(
    apiService: ApiService,
    request: FailedPostRequest,
    cache: FailedRequestCache
): Observable<Response<Any>> {
    return apiService.post(request.url, request.body, request.headers)
        .doOnError { cache.save(request) } // 失败存入缓存
        .doOnNext { cache.remove(request) } // 成功移除缓存
}

// 处理有序请求队列：单个失败不阻塞
fun processOrderedQueue(
    apiService: ApiService,
    requests: List<FailedPostRequest>,
    cache: FailedRequestCache
): Observable<Response<Any>> {
    return Observable.fromIterable(requests)
        .concatMapEager { request ->
            executePost(apiService, request, cache)
                .onErrorResumeNext { Observable.empty() } // 捕获错误，队列继续
        }
}

// 批量并行重发缓存请求：失败不影响其他请求
fun retryCachedRequests(
    apiService: ApiService,
    cache: FailedRequestCache
): Completable {
    val completables = cache.getAll().map { request ->
        apiService.post(request.url, request.body, request.headers)
            .doOnComplete { cache.remove(request) }
            .doOnError { /* 可选：记录错误，或重新缓存（比如网络仍不可用） */ }
            .ignoreElements() // 转为Completable，只关注完成/失败
    }
    return Completable.mergeDelayError(completables)
}

关键注意事项 #

• 错误隔离：必须在单个请求的Observable内部捕获错误，避免错误冒泡到队列流导致整个队列终止。
• 缓存完整性：失败请求的模型要包含重发所需的全部信息（URL、请求体、头部、参数等），确保重发请求与原请求完全一致。
• 重发时机：可以结合网络状态监听（如Android的ConnectivityManager），在网络恢复时自动触发重发；也可以提供手动触发入口。
• 顺序控制：对顺序有要求用concatMapEager，无要求用flatMap+mergeDelayError，平衡顺序性和执行效率。

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