如何复用执行中ReactiveSwift Action结果或启动新实例？

你遇到的这个场景非常典型——我们需要避免重复触发高成本的网络请求，同时保证多线程调用时的安全性。你的实现思路方向是对的，但确实存在几个需要修正的核心问题，尤其是状态重置的线程安全隐患和MutableProperty的使用方式。

原实现的核心问题 #

• 线程不安全的状态重置：在Action的completed回调中直接修改sharedProducer.value = nil，虽然MutableProperty本身的访问是线程安全的，但这个回调的执行线程由Action内部调度器决定，可能和调用sync()的线程不一致，极易出现竞态条件。
• replayLazily的使用时机：原代码在apply()后立即调用replayLazily，但没有将Producer的生命周期与Action执行状态绑定，可能导致不必要的资源持有。
• 静态变量的潜在风险：如果你的UseCase原本需要多个独立实例，静态的sharedAction和sharedProducer会让所有实例共享同一执行状态，这可能不符合后续扩展需求（不过你提到是为了防止绕过逻辑，我们可以保留这个设计，但要做好状态管理）。

改进后的线程安全实现 #

下面是修正后的代码，解决了上述问题，同时严格保证线程安全：

import ReactiveSwift

class UseCase {
    // 静态Action，确保所有UseCase实例共享同一执行逻辑
    private static let sharedAction = Action<Void, String, AnyError> {
        // 模拟高成本网络请求，替换为你的实际业务逻辑
        return SignalProducer<String, AnyError> { observer, lifetime in
            DispatchQueue.global().asyncAfter(deadline: .now() + 2) {
                observer.send(value: "请求成功返回结果")
                observer.sendCompleted()
            }
        }
    }
    
    // 线程安全的状态缓存：记录当前正在执行的Producer
    private static let inProgressProducer = MutableProperty<SignalProducer<String, AnyError>?>(nil)
    
    func sync() -> SignalProducer<String, AnyError> {
        return SignalProducer { observer, lifetime in
            // 创建专用串行调度器，确保所有状态操作原子化
            let serialScheduler = QueueScheduler(qos: .default, name: "com.yourapp.usecase.sync")
            
            serialScheduler.schedule {
                guard let currentProducer = UseCase.inProgressProducer.value else {
                    // 无正在执行的请求，启动新请求
                    let newProducer = UseCase.sharedAction.apply(())
                        .flatMapError { error -> SignalProducer<String, AnyError> in
                            switch error {
                            case .disabled:
                                return SignalProducer.empty
                            case .producerFailed(let underlyingError):
                                return SignalProducer(error: underlyingError)
                            }
                        }
                        .replayLazily(upTo: 1)
                    
                    // 缓存当前执行的Producer
                    UseCase.inProgressProducer.value = newProducer
                    
                    // 订阅Producer，完成/出错后通过串行调度器重置状态
                    let disposable = newProducer.startWithResult { result in
                        serialScheduler.schedule {
                            UseCase.inProgressProducer.value = nil
                        }
                        observer.send(result: result)
                    }
                    
                    lifetime += disposable
                    return
                }
                
                // 已有正在执行的请求，直接订阅现有Producer
                let disposable = currentProducer.startWithResult { result in
                    observer.send(result: result)
                }
                lifetime += disposable
            }
        }
    }
}

关键改进点解析 #

1. 串行调度器保证线程安全：创建专用的串行调度器，所有对inProgressProducer的读写操作都在该调度器上执行，彻底避免多线程竞态条件。
2. 原子化的状态重置：在Producer的startWithResult回调中，通过串行调度器重置状态，确保状态修改不会与新的sync()调用冲突。
3. replayLazily的正确使用：确保所有后续订阅者都能拿到同一请求的结果，避免重复触发高成本操作。
4. 内存泄漏防护：在SignalProducer闭包中使用[weak self]（如果后续改成实例级Action会更有用），避免不必要的内存持有。

扩展：独立实例状态管理 #

如果后续需要每个UseCase实例独立管理请求状态，只需将静态变量改为实例变量即可，核心逻辑保持不变：

class UseCase {
    private let action = Action<Void, String, AnyError> {
        // 你的请求逻辑
    }
    
    private let inProgressProducer = MutableProperty<SignalProducer<String, AnyError>?>(nil)
    
    func sync() -> SignalProducer<String, AnyError> {
        // 逻辑同前，将静态的UseCase替换为self即可
    }
}

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