如何使用 swift combine 异步处理一系列任务

Question

我有一个发布者，它接受网络调用并返回一组 ID。我现在需要为每个 ID 调用另一个网络调用来获取我的所有数据。我希望最终发布者拥有结果对象。

第一个网络结果：

"user": {
   "id": 0,
   "items": [1, 2, 3, 4, 5]
}

最终对象：

struct User {
    let id: Int
    let items: [Item]
    ... other fields ...
}
struct Item {
    let id: Int
    ... other fields ...
}

处理多个网络调用：

userPublisher.flatMap { user in
    let itemIDs = user.items
    return Future<[Item], Never>() { fulfill in
        ... OperationQueue of network requests ...
    }
}

我想并行执行网络请求，因为它们不相互依赖。我不确定 Future 是否就在这里，但我想我会有代码来执行 DispatchGroup 或 OperationQueue 并在它们全部完成时完成。是否有更多的联合方式来做到这一点？

是否有将一个流拆分为多个并行流并将这些流连接在一起的概念？

Answer 1

URLSession除非您确实需要与OperationQueue基于网络的集成，否则结合提供扩展来处理网络请求，然后Future是一个很好的候选者。您可以运行多个Futures 并在某个时候收集它们，但我真的建议查看URLSessionCombine 的扩展。

struct User: Codable {
   var username: String
}

let requestURL = URL(string: "https://example.com/")!
let publisher = URLSession.shared.dataTaskPublisher(for: requestURL)
    .map { $0.data }
    .decode(type: User.self, decoder: JSONDecoder())

关于运行一批请求，可以使用Publishers.MergeMany，即：

struct User: Codable {
   var username: String
}

let userIds = [1, 2, 3]

let subscriber = Just(userIds)
    .setFailureType(to: Error.self)
    .flatMap { (values) -> Publishers.MergeMany<AnyPublisher<User, Error>> in
    let tasks = values.map { (userId) -> AnyPublisher<User, Error> in
            let requestURL = URL(string: "https://jsonplaceholder.typicode.com/users/\(userId)")!

            return URLSession.shared.dataTaskPublisher(for: requestURL)
                .map { $0.data }
                .decode(type: User.self, decoder: JSONDecoder())
                .eraseToAnyPublisher()
    }
    return Publishers.MergeMany(tasks)
}.collect().sink(receiveCompletion: { (completion) in
    if case .failure(let error) = completion {
        print("Got error: \(error.localizedDescription)")
    }
}) { (allUsers) in
    print("Got users:")
    allUsers.map { print("\($0)") }
}

在上面的示例中，我使用collect收集所有结果，这会将值发送到Sink直到所有网络请求成功完成，但是您可以在网络请求完成时删除collect和接收User上面示例中的每一个。