RxSwift如何按顺序执行操作并在每步启停时更新UI

需要按顺序执行刷新令牌、更新目录、拉取数据、插入数据4个同步任务，每个任务启动、执行成功、执行失败时都要更新对应状态，供UI层修改图标颜色。

• 多层concatMap嵌套导致回调地狱，代码可读性差，新增任务需要继续嵌套，维护成本高
• 每个任务的状态更新、日志打印逻辑重复，存在大量冗余代码
• 未统一指定调度器，可能出现后台线程更新UI的异常

1. 统一建模同步任务 #

将所有同步任务封装为枚举，统一关联对应的执行逻辑和状态Relay，避免硬编码每个任务。

2. 通用任务执行包装 #

封装公用方法处理每个任务的状态更新、日志打印逻辑，消除重复代码。

3. 扁平串联任务执行 #

使用Observable.concat按顺序执行任务数组，替代嵌套concatMap，代码结构更清晰。

完整优化代码 #

状态枚举（无需修改）

enum SyncStatus {
    case todo
    case completed
    case error
    case running
    case partial
}

优化后的SyncManager

import RxSwift
import RxCocoa

class SyncManager {
    static let shared = SyncManager()
    private let disposeBag = DisposeBag()
    
    // 各任务状态Relay
    let refreshTokenStatus = BehaviorRelay<SyncStatus>(value: .todo)
    let updateCatalogDataStatus = BehaviorRelay<SyncStatus>(value: .todo)
    let getDataStatus = BehaviorRelay<SyncStatus>(value: .todo)
    let insertDataStatus = BehaviorRelay<SyncStatus>(value: .todo)
    
    // 同步任务枚举：统一管理所有同步任务的配置
    private enum SyncTask {
        case refreshToken
        case updateCatalog
        case getData
        case insertData
        
        // 对应任务的执行流
        var workFlow: Observable<Void> {
            switch self {
            case .refreshToken:
                return RefreshTokenManager.shared.refreshToken().asObservable()
            case .updateCatalog:
                return UpdateCatalogDataSyncManager.shared.updateCatalogData().asObservable()
            case .getData:
                return GetDataSyncManager.shared.getData().asObservable()
            case .insertData:
                return InsertDataWorkSyncManager.shared.insertData().asObservable()
            }
        }
        
        // 对应任务的状态Relay
        var statusRelay: BehaviorRelay<SyncStatus> {
            switch self {
            case .refreshToken: return SyncManager.shared.refreshTokenStatus
            case .updateCatalog: return SyncManager.shared.updateCatalogDataStatus
            case .getData: return SyncManager.shared.getDataStatus
            case .insertData: return SyncManager.shared.insertDataStatus
            }
        }
        
        // 任务名称，用于日志
        var taskName: String {
            switch self {
            case .refreshToken: return "RefreshTokenManager"
            case .updateCatalog: return "UpdateCatalogDataSyncManager"
            case .getData: return "GetDataSyncManager"
            case .insertData: return "InsertDataWorkSyncManager"
            }
        }
    }
    
    private init() {}
    
    func startDatabaseSync(completion: @escaping ((Result<Void, Error>) -> Void)) {
        // 1. 按执行顺序定义所有任务
        let allTasks: [SyncTask] = [.refreshToken, .updateCatalog, .getData, .insertData]
        // 重置所有任务初始状态
        allTasks.forEach { $0.statusRelay.accept(.todo) }
        
        // 2. 包装每个任务，统一处理状态更新和日志
        let taskObservables = allTasks.map { task in
            return task.workFlow
                .do(onSubscribe: {
                    print("\(task.taskName) onSubscribe")
                    task.statusRelay.accept(.running)
                }, onNext: { _ in
                    print("\(task.taskName) onSuccess")
                    task.statusRelay.accept(.completed)
                }, onError: { error in
                    print("\(task.taskName) onError: \(error)")
                    task.statusRelay.accept(.error)
                }, onDispose: {
                    print("\(task.taskName) onDispose")
                })
                .catch { error in
                    // 出错直接抛出，终止后续任务执行，与原有逻辑保持一致
                    Observable.error(error)
                }
        }
        
        // 3. 顺序执行所有任务
        Observable.concat(taskObservables)
            .subscribe(
                onNext: { _ in
                    print("SyncManager onNext")
                },
                onError: { error in
                    print("SyncManager onError: \(error)")
                    completion(.failure(error))
                },
                onCompleted: {
                    print("SyncManager onCompleted")
                    completion(.success(()))
                },
                onDisposed: {
                    print("SyncManager onDisposed")
                }
            )
            .disposed(by: disposeBag)
    }
}

优化后的ViewController监听代码

新增主线程调度控制，避免后台线程更新UI导致的崩溃

SyncManager.shared.refreshTokenStatus
    .observe(on: MainScheduler.instance) // 强制切换到主线程更新UI
    .skip(1)
    .subscribe(onNext: { status in
        // 按状态更新图标颜色：.todo灰色 /.running蓝色 /.completed绿色 /.error红色
    })
    .disposed(by: disposeBag)

SyncManager.shared.updateCatalogDataStatus
    .observe(on: MainScheduler.instance)
    .skip(1)
    .subscribe(onNext: { status in
        // 更新对应UI
    })
    .disposed(by: disposeBag)

SyncManager.shared.insertDataStatus
    .observe(on: MainScheduler.instance)
    .skip(1)
    .subscribe(onNext: { status in
        // 更新对应UI
    })
    .disposed(by: disposeBag)

• 代码结构扁平，没有嵌套回调，可读性大幅提升
• 新增同步任务仅需要扩展SyncTask枚举，不需要修改核心执行逻辑，扩展性更好
• 所有任务的状态更新、日志逻辑统一维护，修改规则不需要逐个调整每个任务的代码
• 统一调度器控制，避免后台线程更新UI导致的崩溃问题

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