swift - "PassthroughSubject"似乎是线程不安全的，这是错误还是限制？

Question

“PassthroughSubject”似乎是线程不安全的。请参阅下面的代码，我同时向订阅者发送 100 个值，该订阅者只请求 .max(5)。订阅者应该只得到 5 个值，我认为，但它实际上得到了更多。这是错误还是限制？

// Xcode11 beta2

var count = 0
let q = DispatchQueue(label: UUID().uuidString)
let g = DispatchGroup()

let subject = PassthroughSubject<Int, Never>()
let subscriber = AnySubscriber<Int, Never>(receiveSubscription: { (s) in
    s.request(.max(5))
}, receiveValue: { v in
    q.sync {
        count += 1
    }
    return .none
}, receiveCompletion: { c in
})
subject.subscribe(subscriber)

for i in 0..<100 {
    DispatchQueue.global().async(group: g) {
        subject.send(i)
    }
}

g.wait()
print("receive", count)  // expected 5, but got more(7, 9...)

Answer 1

我相信前缀 operator可以帮助:

/// Republishes elements up to the specified maximum count.
func prefix(Int) -> Publishers.Output<PassthroughSubject<Output, Failure>>

max 运算符在完成时返回最大值(并且您可能不止一次触发完成):

/// Publishes the maximum value received from the upstream publisher, after it finishes.
/// Available when Output conforms to Comparable.
func max() -> Publishers.Comparison<PassthroughSubject<Output, Failure>>