您可以使用scan将发出的值累积到一个数组中，技巧是在debounce发出该数组后重置该数组：

let subject = PassthroughSubject<Int, Never>()
var reset = false
let cancellable = subject
    .receive(on: RunLoop.main)
    .scan([], { reset ? [$1] : $0 + [$1] })
    .handleEvents(receiveOutput: { _ in reset = false })
    .debounce(for: .seconds(0.5), scheduler: RunLoop.main)
    .handleEvents(receiveOutput: { _ in reset = true })
    .sink { indices in
        print ("Received indices \(indices)")
    }

字符串
但是，这种方法有两个缺陷：

• 您需要将发布服务器切换到主线程
• 外部变量和handleEvents不太“典型”。

不过，您可以将不太好的逻辑 Package 到它自己的发布者中，并且更加习惯：

extension Publishers {
    struct DebouncedCollector<Upstream: Publisher, S: Scheduler>: Publisher {
        typealias Output = [Upstream.Output]
        typealias Failure = Upstream.Failure

        private let upstream: Upstream
        private let dueTime: S.SchedulerTimeType.Stride
        private let scheduler: S
        private let options: S.SchedulerOptions?

        init(upstream: Upstream, dueTime: S.SchedulerTimeType.Stride, scheduler: S, options: S.SchedulerOptions?) {
            self.upstream = upstream
            self.dueTime = dueTime
            self.scheduler = scheduler
            self.options = options
        }

        func receive<S>(subscriber: S) where S : Subscriber, Failure == S.Failure, Output == S.Input {
            var reset = false
            upstream
                .receive(on: scheduler)
                .scan([], { reset ? [$1] : $0 + [$1] })
                .handleEvents(receiveOutput: { _ in reset = false })
                .debounce(for: dueTime, scheduler: scheduler, options: options)
                .handleEvents(receiveOutput: { _ in reset = true })
                .receive(subscriber: subscriber)
        }
    }
}

extension Publisher {
    func collectDebounced<S: Scheduler>(for dueTime: S.SchedulerTimeType.Stride, scheduler: S, options: S.SchedulerOptions? = nil) -> Publishers.DebouncedCollector<Self, S> {
        .init(upstream: self, dueTime: dueTime, scheduler: scheduler, options: options)
    }
}

型
，并像这样使用它：

let subject = PassthroughSubject<Int, Never>()
let cancellable = subject
    .collectDebounced(for: .seconds(0.5), scheduler: RunLoop.main)
    .sink { indices in
        print ("Received indices \(indices)")
    }

型

您不应该使用debounce，因为它是一个过滤操作。相反，使用collect的重载，它接受一个TimeGroupingStrategy-collect * 收集 * 所有来自上游的元素到数组中。

cancellable = subject
    .collect(.byTime(RunLoop.main, 0.5))
    .sink { group in
        print ("Received group \(group)")
    }

字符串

实现这个目标的正确方法是为发布者编写一个自定义操作符，它将对输入值进行反跳，并在所需的延迟之后将它们作为数组传递给下游。我在我的项目中使用它，它工作得很好：

import Foundation
import Combine

struct CollectDebounce<Upstream: Publisher, S: Scheduler>: Publisher {
  typealias Output = [Upstream.Output]
  typealias Failure = Upstream.Failure
  
  let upstream: Upstream
  let dueTime: S.SchedulerTimeType.Stride
  let scheduler: S
  let options: S.SchedulerOptions?
  
  func receive<S>(subscriber: S) where S : Subscriber, Failure == S.Failure, Output == S.Input {
    let debounceSubscriber = CollectDebounceSubscriber(
      downstream: subscriber,
      dueTime: dueTime,
      scheduler: scheduler,
      options: options
    )
    upstream.subscribe(debounceSubscriber)
  }
}

extension CollectDebounce {
  class CollectDebounceSubscriber<Downstream: Subscriber, S: Scheduler>: Subscriber where Downstream.Input == [Upstream.Output], Downstream.Failure == Failure {
    typealias Input = Upstream.Output
    typealias Failure = Downstream.Failure
    
    private let downstream: Downstream
    private let dueTime: S.SchedulerTimeType.Stride
    private let scheduler: S
    private let options: S.SchedulerOptions?
    private var lastCancellable: Cancellable?
    private var collectedValues: [Input] = []
    
    init(downstream: Downstream, dueTime: S.SchedulerTimeType.Stride, scheduler: S, options: S.SchedulerOptions? = nil) {
      self.downstream = downstream
      self.dueTime = dueTime
      self.scheduler = scheduler
      self.options = options
    }
    
    func receive(subscription: Combine.Subscription) {
      downstream.receive(subscription: subscription)
    }
    
    func receive(_ input: Input) -> Subscribers.Demand {
      collectedValues.append(input)
      
      lastCancellable?.cancel()
      lastCancellable = scheduler.schedule(after: scheduler.now.advanced(by: dueTime), interval: .zero, tolerance: .zero) { [weak self] in
        guard let collectedValues = self?.collectedValues else { return }
        _ = self?.downstream.receive(collectedValues)
        self?.collectedValues = []
        self?.lastCancellable?.cancel()
      }
      
      return .none
    }
    
    func receive(completion: Subscribers.Completion<Downstream.Failure>) {
      downstream.receive(completion: completion)
    }
  }
}

extension Publisher {
  func collectDebounce<S: Scheduler>(
    for dueTime: S.SchedulerTimeType.Stride,
    scheduler: S,
    options: S.SchedulerOptions? = nil
  ) -> CollectDebounce<Self, S> {
    return CollectDebounce(upstream: self, dueTime: dueTime, scheduler: scheduler, options: options)
  }
}

字符串
并将其用作出版商的操作员：

yourPublisher
  .collectDebounce(for: .seconds(0.5), scheduler: DispatchQueue.main)
  .sink { array in
    // get array of debounced collection
  }

型