1

“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...)
4

1 回答 1

1

我相信您正在寻找prefix 运营商：

/// 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>>
于 2019-06-24T13:17:48.557 回答