如何将 URLSession.DataTaskPublisher 请求排队，以便一次只发出一个？答案

【问题标题】：How can I queue up URLSession.DataTaskPublisher requests so that only one is made at a time?如何将 URLSession.DataTaskPublisher 请求排队，以便一次只发出一个？
【发布时间】：2020-05-12 13:55:22
【问题描述】：

在下面的代码中，一个app对象数组用于创建一个发布者数组，这些发布者被合并到一个发布对象数组中。

apps.map { latestRelease(app: $0) }.merge()

这是最新版本的完成方式。

func latestRelease(app: App) -> AnyPublisher<Release, Error> {
    do {
        let request = try requestFactory.make(.get, "apps/\(app.owner.name)/\(app.name)/releases/latest")

        return publisherFactory.make(for: request)
            .mapError{ $0 as Error }
            .map { data, _ in data }
            .decode(type: Release.self, decoder: decoder)
            .eraseToAnyPublisher()
    } catch {
        return Fail(error: error)
            .eraseToAnyPublisher()
    }
}

网络请求由工厂完成。

struct AppCenterPublisherFactory: DataTaskPublisherFactory {
    let session: URLSession

    init(session: URLSession = .shared) {
        session.configuration.httpMaximumConnectionsPerHost = 1
        self.session = session
    }

    func make(for request: URLRequest) -> URLSession.DataTaskPublisher {
        return session.dataTaskPublisher(for: request)
    }
}

问题是发布者立即发出网络请求。这会导致服务器返回 429 Too Many Requests。如何将 URLSession.DataTaskPublisher 请求排队，以便一次只发出一个每个请求之间有延迟？

【问题讨论】：

标签： swift combine

【解决方案1】：

如果您使用append 而不是merge 来组合发布者，它们将串行运行而不是同时运行。

对于延迟，您可以将Empty().delay(for: cooldown, scheduler: whatever) 添加到除第一个发布者之外的每个发布者。

func releases<S: Scheduler>(of apps: [App], scheduler: S, cooldown: S.SchedulerTimeType.Stride)
    -> AnyPublisher<Release, Error>
{
    let singles = apps.map { latestRelease(app: $0) }
    guard let first = singles.first else { return Empty().eraseToAnyPublisher() }

    let combo: AnyPublisher<Release, Error> = singles.dropFirst()
        .reduce(first.eraseToAnyPublisher(), { combo, single in
            combo
                .append(Empty().delay(for: cooldown, scheduler: scheduler))
                .append(single)
                .eraseToAnyPublisher()
        })

    return combo
}

测试：

let testApps: [App] = [
    .init(name: "Facebook", owner: .init(name: "zuck")),
    .init(name: "Kindle", owner: .init(name: "bezos")),
    .init(name: "Crossword", owner: .init(name: "shortz")),
]

print("starting at \(Date())")
let ticket = releases(of: testApps, scheduler: DispatchQueue.global(qos: .utility), cooldown: .seconds(2))
    .sink(
        receiveCompletion: { print("got \($0) at \(Date())") },
        receiveValue: { print("got \($0) at \(Date())") })

输出：

starting at 2020-05-12 17:45:17 +0000
got Release(name: "Facebook") at 2020-05-12 17:45:17 +0000
got Release(name: "Kindle") at 2020-05-12 17:45:19 +0000
got Release(name: "Crossword") at 2020-05-12 17:45:21 +0000
got finished at 2020-05-12 17:45:21 +0000

【讨论】：

这是完美的答案。非常感谢。但是，我不得不想出一个不同的解决方案。事实证明，拥有数百个发布者的数组并在所有发布者上调用 append() 会导致 Swift 5.1.3 崩溃。

【解决方案2】：

您可以使用 receive(on:options:) 在指定的调度程序上交付任务。作为样本

let queue = DispatchQueue(label: "App_Queue", qos: .default)

那就这样改

publisherFactory.make(for: request).receive(on: queue) // rest of code...

希望你能绕过它。谢谢，X_X

【讨论】：

我建议不要像你建议的那样在latestRelease 函数中添加receive(on:)，而是在map 中调用它，将latestRelease 本身分派到串行队列。 apps.map { latestRelease(app: $0).receive(on: queue) }.merge()
如何延迟每个请求？