Swift: 如何使用Combine执行并发API调用

Question

我想利用Combine框架进行并发API调用。这些API调用设置如下：

1. 首先，调用一个API获取帖子列表。
2. 对于每个帖子，调用另一个API以获取评论。

我希望使用Combine将这两个调用链接在一起，并同时执行，以便返回一个包含每篇帖子及其评论数组的帖子对象数组。

我的尝试：

struct Post: Decodable {
    let userId: Int
    let id: Int
    let title: String
    let body: String
    var comments: [Comment]?
}

struct Comment: Decodable {
    let postId: Int
    let id: Int
    let name: String
    let email: String
    let body: String
}

class APIClient: ObservableObject {
    @Published var posts = [Post]()
    
    var cancellables = Set<AnyCancellable>()
    
    init() {
        getPosts()
    }
    
    func getPosts() {
        let urlString = "https://jsonplaceholder.typicode.com/posts"
        guard let url = URL(string: urlString) else {return}
        
        URLSession.shared.dataTaskPublisher(for: url)
            .receive(on: DispatchQueue.main)
            .tryMap({ (data, response) -> Data in
                guard
                    let response = response as? HTTPURLResponse,
                    response.statusCode >= 200 else {
                    throw URLError(.badServerResponse)
                }
                
                return data
            })
            .decode(type: [Post].self, decoder: JSONDecoder())
            .sink { (completion) in
                print("Posts completed: \(completion)")
            } receiveValue: { (output) in
                //Is there a way to chain getComments such that receiveValue would contain Comments??
                output.forEach { (post) in
                    self.getComments(post: post)
                }
            }
            .store(in: &cancellables)
    }
    
    func getComments(post: Post) {
        let urlString = "https://jsonplaceholder.typicode.com/posts/\(post.id)/comments"
        guard let url = URL(string: urlString) else {
            return
        }
        
        URLSession.shared.dataTaskPublisher(for: url)
            .receive(on: DispatchQueue.main)
            .tryMap({ (data, response) -> Data in
                guard
                    let response = response as? HTTPURLResponse,
                    response.statusCode >= 200 else {
                    throw URLError(.badServerResponse)
                }
                
                return data
            })
            .decode(type: [Comment].self, decoder: JSONDecoder())
            .sink { (completion) in
                print("Comments completed: \(completion)")
            } receiveValue: { (output) in
                print("Comment", output)
            }
            .store(in: &cancellables)
    }
}

如何将getComments链接到getPosts，以便可以在getPosts中接收评论输出？传统上，使用UIKit，我会使用DispatchGroup来完成此任务。

请注意，我希望从APIClient仅接收一次发布事件的文章，以便SwiftUI视图只刷新一次。

- Koh

---

在编程中，是否有像rxjs中的CombineLatest这样的操作符？ - Dan Chase

@DanChase 有一个 CombineLatest 方法，但我想这个方法会监听两个发布者。你有什么想法如何在上述用例中应用它吗？ - Koh

https://dev59.com/VlIH5IYBdhLWcg3wAHYv - matt

@Koh 抱歉我误读了使用案例。这不是一个实际的答案，但在过去，我通过在后端创建一个组合结构并只在HTTP端调用一次来解决了这个问题。在后端，API表面可以调用多个业务层函数并创建一个结构来返回。在我目前正在工作的一个项目中，它已经增长到了8个HTTP Get请求，并且我开始遇到一些返回比其他请求更早的问题，这会给用户带来困惑，同时也会导致浏览器出现问题。我相信HTTP 1.1有一个6的限制...希望这比我的上一个评论更有帮助。 - Dan Chase

@Koh，我再考虑一下combineLatest，我认为我的思路是使用两个可观察对象与每个循环配对，一个用于主要内容，另一个用于详细信息。但是，我越想越觉得这个想法越来越糟糕，这也导致了我上面的评论。 - Dan Chase

都很好，但现在我们有了Swift-5.5，“Swift内置支持以结构化方式编写异步和并行代码。”我一定会看看它，特别是使用actors。https://docs.swift.org/swift-book/LanguageGuide/Concurrency.html网络上有一些很好的教程。 - workingdog support Ukraine

Answer 1

感谢@matt在上面的评论中的帖子，我已经针对上述用例调整了那个SO帖子中的解决方案。

我不太确定这是否是最佳实现，但它暂时解决了我的问题。

  func getPosts() {
        let urlString = "https://jsonplaceholder.typicode.com/posts"
        guard let url = URL(string: urlString) else {return}
        
        URLSession.shared.dataTaskPublisher(for: url)
            .receive(on: DispatchQueue.main)
            .tryMap({ (data, response) -> Data in
                guard
                    let response = response as? HTTPURLResponse,
                    response.statusCode >= 200 else {
                    throw URLError(.badServerResponse)
                }
                
                return data
            })
            .decode(type: [Post].self, decoder: JSONDecoder())
            .flatMap({ (posts) -> AnyPublisher<Post, Error> in
                //Because we return an array of Post in decode(), we need to convert it into an array of publishers but broadcast as 1 publisher
                Publishers.Sequence(sequence: posts).eraseToAnyPublisher()
            })
            .compactMap({ post in
                //Loop over each post and map to a Publisher
                self.getComments(post: post) 
            })
            .flatMap {$0} //Receives the first element, ie the Post
            .collect() //Consolidates into an array of Posts
            .sink(receiveCompletion: { (completion) in
                print("Completion:", completion)
            }, receiveValue: { (posts) in
                self.posts = posts
            })
            .store(in: &cancellables)
    }
    
    func getComments(post: Post) -> AnyPublisher<Post, Error>? {
        let urlString = "https://jsonplaceholder.typicode.com/posts/\(post.id)/comments"
        guard let url = URL(string: urlString) else {
            return nil
        }
        
        let publisher = URLSession.shared.dataTaskPublisher(for: url)
            .receive(on: DispatchQueue.main)
            .tryMap({ (data, response) -> Data in
                guard
                    let response = response as? HTTPURLResponse,
                    response.statusCode >= 200 else {
                    throw URLError(.badServerResponse)
                }

                return data
            })
            .decode(type: [Comment].self, decoder: JSONDecoder())
            .tryMap { (comments) -> Post in
                var newPost = post
                newPost.comments = comments
                return newPost
            }
            .eraseToAnyPublisher()
        
        return publisher
    }

基本上，我们需要从getComments方法返回一个Publisher，以便我们可以在getPosts中循环遍历每个publisher。

- Koh