如何使用 Combine 限制并发实时 URLSession？答案

【问题标题】：How to limit concurrent live URLSessions with Combine?如何使用 Combine 限制并发实时 URLSession？
【发布时间】：2022-01-02 09:29:59
【问题描述】：

我有很多（约 200 个）图片 url，我需要下载每一个，然后处理（调整大小）它，然后更新缓存。问题是 - 我只想一次最多有 3 个请求，并且由于图像很重，我也不希望有很多响应“挂起”等待处理（并占用内存......）。

TLDR 我只想在前 3 个请求之一调用 sink 中的 receiveValue 之后调用下一个（第 4 个）网络请求......（即在网络响应和处理都完成之后。 ..)。

这个流程会起作用吗，它会保留等待的 url 而不会把它们丢在地板上吗？

我还需要buffer() 电话吗？我看到这个答案后使用它：https://stackoverflow.com/a/67011837/2242359

wayTooManyURLsToHandleAtOnce // this is a `[URL]`
    .publisher
    .buffer(size: .max, prefetch: .byRequest, whenFull: .dropNewest) // NEEDED?
    .flatMap(maxPublishers: .max(3)) { url in
       URLSession.shared
           .dataTaskPublisher(for: url)
           .map { (data: Data, _) -> Picture in
               Picture(from: data)
           }
    }
    .tryCompactMap {
        resizeImage(picture: $0) // takes a while and might fail
    }
    .receive(on: DispatchQueue.main)
    .sink { completion
        // handling completion... 
    } receiveValue: { resizedImage
        self.cache.append(resizedImage)
    }
    .store(...)

【问题讨论】：

如果您对您的代码是否有效或是否需要缓冲区有任何疑问，为什么不编写一个测试来练习您的策略并找出答案？只需将您自己的发布者替换为 URLSession 发布者，这样就不会涉及网络并且您可以控制时间。无需猜测！
是的，我做了一些测试，它似乎按预期工作，但由于我缺乏 API 经验，我仍然害怕它“偶然工作”，而且我也不确定II 是以“正确的方式”做的，或者是否有更好的工具来完成这项工作
对我来说看起来不错，只是我有点担心你在哪里做resizeImage。但这不是你的问题。 — 请注意，当我说“测试”时，我说的是在没有网络的情况下进行测试。
我确实在离线时得到了类似的反馈（：基本上反馈是调整大小应该是 flatMap 内发布者链的一部分，包括在接收器中完成的缓存附加，接收器应该只通知 UI。这是你的想法吗？
是的。如果没有必要，您不应该在直接链中做“耗时且可能出错”的事情。

标签： swift combine urlsession

【解决方案1】：

我会使用一个主题。这不是最佳解决方案，但它看起来很有效，可能会引发其他想法

var cancellable: AnyCancellable?

var urls: [String] = (0...6).map { _ in "http://httpbin.org/delay/" + String((0...2).randomElement()!) }

var subject: PassthroughSubject<[String], Never> = .init()

let maxConcurrentRequests = 3

override func viewDidAppear(_ animated: Bool) {
    super.viewDidAppear(animated)
    
    print(urls)
    
    cancellable = subject
        .flatMap({ urls -> AnyPublisher<[URLSession.DataTaskPublisher.Output], URLError> in
            let requests = urls.map { URLSession.shared.dataTaskPublisher(for: URL.init(string: $0)!) }
            return Publishers.MergeMany(requests)
                .collect().eraseToAnyPublisher()
        })
        .print()
        .sink(receiveCompletion: { completion in
            print(completion)
        }, receiveValue: { value in
            print(value)
            if self.urls.count <= self.maxConcurrentRequests {
                self.urls.removeAll()
                self.subject.send(completion: .finished)
            } else {
                self.urls.removeLast(self.maxConcurrentRequests)
                self.subject.send(self.urls.suffix(self.maxConcurrentRequests))
            }
        })
    
    subject.send(urls.suffix(maxConcurrentRequests))
}

【讨论】：

不会在接收器上调用self.subject.send(completion: .finished) 永远结束我的订阅吗？（即忽略发出的未来值）
@AvielGross 是的，确实如此。我知道您的 url 集合为每个视图控制器创建一次/关闭。如果它不是真的，那么不要发送“.finished”，但是你有一个触发器，比如 urls 数组上的 didSet 或另一个触发器，在清空数组并重新填充它之后用 subject.send 重新启动。