RxSwift RetryWhen 导致重入异常

Question

我一直在尝试对retryWhen运算符进行测试RxSwift，但遇到了Reentrancy Anomaly问题，代码如下：

\n\n

Observable<Int>.create { observer in\n    observer.onNext(1)\n    observer.onNext(2)\n    observer.onNext(3)\n    observer.onNext(4)\n    observer.onError(RequestError.dataError)\n    return Disposables.create()\n    }\n    .retryWhen { error in\n        return error.enumerated().flatMap { (index, error) -> Observable<Int> in\n        let maxRetry = 1\n        print("index: \\(index)")\n        return index < maxRetry ? Observable.timer(1, scheduler: MainScheduler.instance) : Observable.error(RequestError.tooMany)\n        }\n    }\n    .subscribe(onNext: { value in\n        print("This: \\(value)")\n    }, onError: { error in\n        print("ERRRRRRR: \\(error)")\n    })\n    .disposed(by: disposeBag)\n

\n\n

通过上面的代码，它给出：

\n\n

This: 1\nThis: 2\nThis: 3\nThis: 4\nindex: 0\nThis: 1\nThis: 2\nThis: 3\nThis: 4\nindex: 1\n\xe2\x9a\xa0\xef\xb8\x8f Reentrancy anomaly was detected.\n  > Debugging: To debug this issue you can set a breakpoint in /Users/tony.lin/Documents/Snippet/MaterialiseTest/Pods/RxSwift/RxSwift/Rx.swift:97 and observe the call stack.\n  > Problem: This behavior is breaking the observable sequence grammar. `next (error | completed)?`\n    This behavior breaks the grammar because there is overlapping between sequence events.\n    Observable sequence is trying to send an event before sending of previous event has finished.\n  > Interpretation: This could mean that there is some kind of unexpected cyclic dependency in your code,\n    or that the system is not behaving in the expected way.\n  > Remedy: If this is the expected behavior this message can be suppressed by adding `.observeOn(MainScheduler.asyncInstance)`\n    or by enqueing sequence events in some other way.\n\n\xe2\x9a\xa0\xef\xb8\x8f Reentrancy anomaly was detected.\n  > Debugging: To debug this issue you can set a breakpoint in /Users/tony.lin/Documents/Snippet/MaterialiseTest/Pods/RxSwift/RxSwift/Rx.swift:97 and observe the call stack.\n  > Problem: This behavior is breaking the observable sequence grammar. `next (error | completed)?`\n    This behavior breaks the grammar because there is overlapping between sequence events.\n    Observable sequence is trying to send an event before sending of previous event has finished.\n  > Interpretation: This could mean that there is some kind of unexpected cyclic dependency in your code,\n    or that the system is not behaving in the expected way.\n  > Remedy: If this is the expected behavior this message can be suppressed by adding `.observeOn(MainScheduler.asyncInstance)`\n    or by enqueing sequence events in some other way.\n\nERRRRRRR: tooMany\n

\n\n

只是想知道是否有人知道这个问题的原因？

\n

Answer 1

正如控制台注释所解释的，可以使用.observeOn(MainScheduler.asyncInstance)以下命令来抑制此警告：

Observable<Int>.from([1, 2, 3, 4]).concat(Observable.error(RequestError.dataError))
    .observeOn(MainScheduler.asyncInstance) // this is the magic that makes it work.
    .retryWhen { error in
        return error.enumerated().flatMap { (index, error) -> Observable<Int> in
            let maxRetry = 1
            print("Index:", index)
            guard index < maxRetry else { throw RequestError.tooMany }
            return Observable.timer(1, scheduler: MainScheduler.instance)
        }
    }
    .subscribe(onNext: { value in
        print("This: \(value)")
    }, onError: { error in
        print("ERRRRRRR: \(error)")
    })

我冒昧地对您的示例代码进行了一些细微的调整，以展示编写您所拥有的内容的另一种方法。

附加信息

您要求解释 (a) 为什么添加 ObserveOn 有效以及 (b) 为什么需要它。

所做.observeOn(MainScheduler.asyncInstance)的是将请求路由到备用线程，事件可以在其中完成，然后在主线程上再次发出事件。换句话说，就像这样做：

.observeOn(backgroundScheduler).observeOn(MainScheduler.instance)

其中backgroundScheduler定义如下：

let backgroundScheduler = SerialDispatchQueueScheduler(qos: .default)

至少这是我的理解。

至于为什么需要它，我也说不上来。您可能在库中发现了一个错误，因为在没有observeOn的情况下使用1秒延迟也可以正常工作。