使用合并，如何在网络请求后取消分配订阅

Question

如果将URLSession用于网络请求的合并，则需要保存Subscription（又名AnyCancellable）-否则它将立即被释放，这将取消网络请求。稍后，在处理完网络响应后，您要取消分配订阅，因为保留该订阅将浪费内存。

下面是一些执行此操作的代码。这有点尴尬，甚至可能不正确。我可以想象在sub设置为非nil值之前网络请求可以在另一个线程上启动并完成的竞争状态。

有更好的方法吗？

class SomeThing {
    var subs = Set<AnyCancellable>()
    func sendNetworkRequest() {
        var request: URLRequest = ...
        var sub: AnyCancellable? = nil            
        sub = URLSession.shared.dataTaskPublisher(for: request)
            .map(\.data)
            .decode(type: MyResponse.self, decoder: JSONDecoder())
            .sink(
                receiveCompletion: { completion in                
                    self.subs.remove(sub!)
                }, 
                receiveValue: { response in ... }
            }
        subs.insert(sub!)

Answer 1

下面是一些执行此操作的代码。这有点尴尬，甚至可能不正确。我可以想象一个竞争条件，在将sub设置为非nil值之前，网络请求可以在另一个线程上启动并完成。

危险！ Swift.Set不是线程安全的。如果要从两个不同的线程访问Set，则要由您序列化访问，以使它们不重叠。

[通常可能（尽管URLSession.DataTaskPublisher可能不是）是在sink运算符返回之前，发布者同步发出其信号。这就是Just，Result.Publisher，Publishers.Sequence等的行为。因此，这些都会产生您正在描述的问题，而不会涉及线程安全性。

现在，如何解决问题？如果您不希望实际取消订阅，则可以完全避免使用AnyCancellable而不是Subscribers.Sink运算符来创建sink：

        URLSession.shared.dataTaskPublisher(for: request)
            .map(\.data)
            .decode(type: MyResponse.self, decoder: JSONDecoder())
            .subscribe(Subscribers.Sink(
                receiveCompletion: { completion in ... },
                receiveValue: { response in ... }
            ))

在订阅完成后（使用.finished或.failure），Combine将清除订阅和订阅者。

但是，如果您
do
希望能够取消订阅该怎么办？有时您的SomeThing在订阅完成之前就被销毁了，在这种情况下，您不需要完成它。因此，您确实需要创建和存储AnyCancellable，并且回到了竞争状态。
在那种情况下，设置一个指示接收器赢得比赛的标志，并在存储AnyCancellable之前检查该标志。
        var sub: AnyCancellable? = nil
        var isComplete = false
        sub = URLSession.shared.dataTaskPublisher(for: request)
            .map(\.data)
            .decode(type: MyResponse.self, decoder: JSONDecoder())
            // This ensures thread safety, if the subscription is also created
            // on DispatchQueue.main.
            .receive(on: DispatchQueue.main)
            .sink(
                receiveCompletion: { [weak self] completion in
                    isComplete = true
                    if let theSub = sub {
                        self?.subs.remove(theSub)
                    }
                }, 
                receiveValue: { response in ... }
            }
        if !isComplete {
            subs.insert(sub!)
        }