RxSwift中如何动态添加Observable任务并按FIFO顺序串行执行？

核心思路是用PublishSubject接收动态提交的任务，通过限制最大并发数为1的操作符保证任务FIFO串行执行，前一个任务完全执行结束后才会启动下一个任务。

具体实现 #

你需要封装一个独立的任务管理类，避免队列状态混乱：

import RxSwift

class ChunkModifyManager {
    // 存储待执行任务的Subject，存任务构造闭包避免任务提前触发执行
    private let taskSubject = PublishSubject<() -> Observable<Double>>()
    private let disposeBag = DisposeBag()
    
    init() {
        bindTaskExecutionLogic()
    }
    
    private func bindTaskExecutionLogic() {
        taskSubject
            // 限制最大并发数为1，天然保证FIFO串行执行
            .flatMap(.concurrent(max: 1)) { taskBuilder in
                // 捕获单个任务的错误，避免单任务失败导致整个队列终止
                taskBuilder()
                    .catch { error in
                        // 可在此处添加全局错误处理逻辑
                        print("任务执行失败: \(error.localizedDescription)")
                        return .empty()
                    }
            }
            .subscribe()
            .disposed(by: disposeBag)
    }
    
    // 对外暴露的任务提交方法
    func submitTask(data: SomeData) -> Observable<Double> {
        return Observable.create { [weak self] observer in
            guard let self = self else {
                observer.onCompleted()
                return Disposables.create()
            }
            // 构造任务闭包，等到队列调度到该任务时才会生成Observable启动执行
            let task = { modifyChunk(data: data) }
            // 把任务的执行事件转发给调用方，支持单独监听单个任务的进度
            let taskDisposable = task().subscribe(observer)
            // 提交任务到队列
            self.taskSubject.onNext(task)
            return taskDisposable
        }
    }
}

逻辑说明 #

• 任务用闭包包装的原因：modifyChunk返回的是冷Observable，一旦生成并被订阅就会立即执行，用闭包包装后只有队列调度到该任务时才会调用闭包生成Observable，避免任务提前执行打乱顺序。
• flatMap(.concurrent(max: 1))是核心：该操作符会保证同一时间最多只有一个内部Observable处于运行状态，前一个任务发送completed事件后才会订阅下一个任务的Observable，完全满足FIFO执行要求。
• 错误兼容：单个任务执行失败不会终止整个队列，后续提交的任务依然可以正常运行。

使用示例 #

// 全局持有管理类实例
let modifyManager = ChunkModifyManager()
let disposeBag = DisposeBag()

// 用户发起第一次修改
modifyManager.submitTask(data: dataA)
    .subscribe(onNext: { progress in
        print("任务A执行进度：\(progress)")
    })
    .disposed(by: disposeBag)

// 用户发起第二次修改，会自动等待任务A执行完成后再启动
modifyManager.submitTask(data: dataB)
    .subscribe(onNext: { progress in
        print("任务B执行进度：\(progress)")
    })
    .disposed(by: disposeBag)

注意事项 #

• 取消单个任务时，只需要销毁submitTask返回的订阅即可，当前任务终止后队列会自动执行下一个任务。
• 如果需要清空队列、暂停/恢复执行，可以扩展管理类添加对应的控制逻辑，比如用filter操作符控制队列是否执行任务，或者重新初始化taskSubject清空待执行任务。

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