合并没有现成的解决方案，但我们可以在现有的Publishers.MergeMany和Publishers.Concatenate上构建它。
其理念是：

• 将输入数组划分为最大并发请求的块。例如，使用简单的Int数组[1, 2, 3, 4, 5, 6]和maxConcurrent = 3，我们将得到[[1, 2, 3], [4, 5, 6]]，因此请求1，2，3将并行执行，但4，5，6将仅在前一个块完成时开始。
• 在这些子数组上使用Publishers.MergeMany。
• 将上一步骤中产生的每个新发布者连接起来。

为了实现这一点，我们需要利用Publishers.Concatenate来实现Publishers.ConcatenateMany，它只需要2个输入流。如果你想遵循合并风格，这应该在一个全新的结构中实现，但我现在在静态函数中实现了这一点。

extension Publishers {
  static func concatenateMany<Output, Failure>(_ publishers: [AnyPublisher<Output, Failure>]) -> AnyPublisher<Output, Failure> {
    return publishers.reduce(Empty().eraseToAnyPublisher()) { acc, elem in
      Publishers.Concatenate(prefix: acc, suffix: elem).eraseToAnyPublisher()
    }
  }

用于分块数组的实用程序：

extension Array {
  func chunked(into size: Int) -> [[Element]] {
    return stride(from: 0, to: count, by: size).map {
      Array(self[$0 ..< Swift.min($0 + size, count)])
    }
  }
}

现在我们可以实现一个新版本的MergeMany，它也接受一个maxConcurrent参数。

extension Publishers {
  static func mergeMany<Output, Failure>(maxConcurrent: Int, _ publishers: [AnyPublisher<Output, Failure>]) -> AnyPublisher<Output, Failure> {
    return Publishers.concatenateMany(
      publishers.chunked(into: maxConcurrent)
        .map {
          Publishers.MergeMany($0)
            .eraseToAnyPublisher()
        }
    )
  }
}

最后，你的代码看起来像这样：

let mergedPubs = Publishers.mergeMany(maxConcurrent: 3, requests)
  .collect()
  .eraseToAnyPublisher()

这只是一个想法，可能还有其他方法可以达到同样的效果！

虽然@Fabio Felici有一个很酷的并行化工作解决方案，但在这个特定的情况下，似乎你有一个错误没有在这里显示，而解决方案不在你的代码的这一部分。
因此，在这种情况下，每次调用方法dataTaskPublisher(for: $0)时，看起来就像是在创建一个新的会话。这使得您只受系统资源的限制，可以发出多少请求。这里的解决方案是保留一个对'URLSession'的引用，并从该单个示例创建数据任务发布者。创建会话时，你应该向会话的初始化器传递一个配置，在那个配置中，将httpMaximumConnectionsPerHost设置为你想要建立的最大并发连接数。你可以这样做：

struct RawJSON: Decodable {
    /// properties
}

class YourClass {
    
    let session = {
        /// Create a `URLSessionConfiguration`
        let configuration = URLSessionConfiguration()
        /// Set the maximum concurrent connections you want
        configuration.httpMaximumConnectionsPerHost = 10
        return URLSession(configuration: configuration, delegate: nil, delegateQueue: nil)
    }()

    /// I am assuming this is what your `dataTaskPublisher` method looks like
    func dataTaskPublisher(for url: URL) -> URLSession.DataTaskPublisher {
        return session.dataTaskPublisher(for: url)
    }

    /// This is your method that you posted here
    func yourMethod(urlRequests: [URL]) {
        let mergedPubs = Publishers.MergeMany(urlRequests.map { dataTaskPublisher(for: $0)
                    .decode(type: RawJSON.self, decoder: JSONDecoder())
                    .mapError { _ in
                        return URLError(URLError.Code.badServerResponse)
                    }
            })
                .collect()
                .eraseToAnyPublisher()
    }
}

如果你想更聪明一点，你可以根据你选择的算法动态地改变并发操作的数量。

struct RawJSON: Decodable {
    /// properties
}

class YourClass {
    
    let operationQueue: OperationQueue
    let session: URLSession
    init() {
        /// Create a `URLSessionConfiguration`
        let configuration = URLSessionConfiguration()
        /// Create `OperationQueue`
        operationQueue = OperationQueue()
        /// Set the maximum concurrent connections you want
        configuration.httpMaximumConnectionsPerHost = 10
        session = URLSession(configuration: configuration, delegate: nil, delegateQueue: nil)
    }

    /// I am assuming this is what your `dataTaskPublisher` method looks like
    func dataTaskPublisher(for url: URL) -> URLSession.DataTaskPublisher {
        return session.dataTaskPublisher(for: url)
    }

    /// This is your method that you posted here.
    func yourMethod(urlRequests: [URL]) {
        let mergedPubs = Publishers.MergeMany(urlRequests.map { dataTaskPublisher(for: $0)
                    .decode(type: RawJSON.self, decoder: JSONDecoder())
                    .mapError { _ in
                        return URLError(URLError.Code.badServerResponse)
                    }
            })
                .collect()
                .eraseToAnyPublisher()
    }
    
    /// Call this to change the number of concurrent operations in the queue
    func changeCuncurrentOperationCount(maxOperations: Int) {
        operationQueue.maxConcurrentOperationCount = maxOperations
    }
}

你可以这样使用它：