ios - publishSubject 同步异常警告

Question

我在 Playground 上玩 RxSwift 时遇到了警告。 这是完整的警告消息:

Synchronization anomaly was detected.
- Debugging: To debug this issue you can set a breakpoint in RxSwift/RxSwift/Rx.swift:113 and observe the call stack.
Problem: This behavior is breaking the observable sequence grammar. `next (error | completed)?`
- This behavior breaks the grammar because there is overlapping between sequence events.
Observable sequence is trying to send an event before sending of previous event has finished.
- Interpretation: Two different unsynchronized threads are trying to send some event simultaneously.
This is undefined behavior because the ordering of the effects caused by these events is nondeterministic and depends on the 
operating system thread scheduler. This will result in a random behavior of your program.
- Remedy: If this is the expected behavior this message can be suppressed by adding `.observeOn(MainScheduler.asyncInstance)`
or by synchronizing sequence events in some other way.

这是 Playground 上的代码。

import RxSwift
import Foundation

example("PublishSubject") {
let disposeBag = DisposeBag()
let subject = PublishSubject<String>()


subject.onNext("🐶")
subject.subscribe(onNext: { (value) in
    print(value)
}).disposed(by: disposeBag)
subject.onNext("🐱")

subject.onNext("🅰️")
subject.onNext("🅱️")
DispatchQueue.global(qos: .utility).async {
    for index in 0...10 {
        subject.onNext("1")
    }
          subject.observeOn(MainScheduler.asyncInstance).subscribe(onNext: { (value) in
        print(value)
    }).disposed(by: disposeBag)
}
DispatchQueue.global(qos: .utility).async {
    for index in 0...10 {
        subject.onNext("B")
    }
    subject.observeOn(MainScheduler.asyncInstance).subscribe(onNext: { (value) in
        print(value)
    }).disposed(by: disposeBag)
}
}

我该怎么做才能解决这个警告问题？？谢谢

Answer 1

当主题正在处理事件时，您正在发送关于主题的事件。这打破了受试者必须维持的契约。

具体来说，主题中没有任何线程跟踪，因此您必须手动进行。最明显的方法是在您的 onNext 调用周围放置一个递归锁，这样它们在单独的线程上运行时就不会重叠。

let disposeBag = DisposeBag()
let subject = PublishSubject<String>()
let lock = NSRecursiveLock()

subject.onNext("🐶")
subject.subscribe(onNext: { (value) in
    print(value)
}).disposed(by: disposeBag)
subject.onNext("🐱")

subject.onNext("🅰️")
subject.onNext("🅱️")
DispatchQueue.global(qos: .utility).async {
    for index in 0...10 {
        lock.lock()
        subject.onNext("1")
        lock.unlock()
    }
    subject.observeOn(MainScheduler.asyncInstance).subscribe(onNext: { (value) in
        print(value)
    }).disposed(by: disposeBag)
}
DispatchQueue.global(qos: .utility).async {
    for index in 0...10 {
        lock.lock()
        subject.onNext("B")
        lock.unlock()
    }
    subject.observeOn(MainScheduler.asyncInstance).subscribe(onNext: { (value) in
        print(value)
    }).disposed(by: disposeBag)
}