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

Question

如果使用 URLSession 对网络请求使用组合，则需要保存 Subscription（又名 AnyCancellable） - 否则会立即解除分配，从而取消网络请求。稍后，当网络响应处理完毕后，您希望取消分配订阅，因为保留它会浪费内存。

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

有更好的方法吗？

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

我将这种情况称为一次性订阅者。这个想法是，因为数据任务发布者只发布一次，所以您知道在收到单个值和/或完成（错误）后销毁管道是安全的。

这是我喜欢使用的一种技术。首先，这是管道的负责人：

let url = URL(string:"https://www.apeth.com/pep/manny.jpg")!
let pub : AnyPublisher<UIImage?,Never> =
    URLSession.shared.dataTaskPublisher(for: url)
        .map {$0.data}
        .replaceError(with: Data())
        .compactMap { UIImage(data:$0) }
        .receive(on: DispatchQueue.main)
        .eraseToAnyPublisher()

现在是有趣的部分。仔细观察：

var cancellable: AnyCancellable? // 1
cancellable = pub.sink(receiveCompletion: {_ in // 2
    cancellable?.cancel() // 3
}) { image in
    self.imageView.image = image
}

你看到我在那里做了什么吗？也许不是，所以我会解释一下：

1. 首先，我声明一个local AnyCancellable 变量；由于与 Swift 语法规则有关的原因，这需要是一个 Optional。

2. 然后，我创建我的订阅者并将我的 AnyCancellable 变量设置为该订阅者。同样，由于与 Swift 语法规则有关的原因，我的订阅者需要是一个 Sink。

3. 最后，在订阅者本身，我在收到完成后取消 AnyCancellable。

第三步中的取消实际上做了两件事，除了调用cancel()——与内存管理有关的事情：

• 通过引用到 Sink 的异步完成函数内的cancellable，我将cancellable 和整个管道 活动 保持足够长的时间以获取值从订阅者那里到达。

• 通过取消cancellable，我允许管道不存在并防止可能导致周围视图控制器泄漏的保留周期。

Answer 2

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

危险！ Swift.Set 不是线程安全的。如果您想从两个不同的线程访问Set，则由您决定对访问进行序列化，以免它们重叠。

一般情况下（虽然URLSession.DataTaskPublisher 可能不行）是发布者在sink 操作符返回之前同步发出其信号。这就是Just、Result.Publisher、Publishers.Sequence 和其他人的行为方式。所以这些会产生你所描述的问题，而不涉及线程安全。

现在，如何解决这个问题？如果您认为您实际上并不想取消订阅，那么您可以通过使用 Subscribers.Sink 而不是 sink 运算符来完全避免创建 AnyCancellable：

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

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

但是如果您确实希望能够取消订阅怎么办？也许有时您的SomeThing 在订阅完成之前就被销毁了，在这种情况下您不需要完成订阅。然后你确实想创建一个AnyCancellable 并将它存储在一个实例属性中，这样当SomeThing 被销毁时它就会被取消。

在这种情况下，设置一个标志表明 sink 赢得了比赛，并在存储 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!)
        }

Answer 3

组合发布者有一个名为 prefix 的实例方法，它执行以下操作：

func prefix(_ maxLength: Int) -> Publishers.Output<Self>

https://developer.apple.com/documentation/combine/publisher/prefix(_:)

playground example