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

Question

我有很多(~200(个图像url，我需要下载每个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(...)

Answer 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))
}