发布主题同步异常警告

Question

我在操场上玩 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)
}