调用onBackpressureBlock()后甚至出现MissingBackpressureException

Question

我试图定期发出事件(每150ms),即使上游observable将更快地发送事件.

但MissingBackpressureException即使我打电话,我也会得到onBackpressureBlock()

码:

    SerializedSubject<QuotationMarker, QuotationMarker> subject = new SerializedSubject<> (PublishSubject.create());

    return subject
            .subscribeOn(Schedulers.computation())
            .doOnSubscribe(() -> {
                NetworkRequestsManager.instance().queryQuotations(productId).subscribe(quotation -> {
                            Log.d(TAG, "new quotation " + quotation.hashCode());
                            NetworkRequestsManager.instance().getSeller(quotation.sellerId)
                                    .subscribe(seller -> {
                                                for (Outlet outlet : seller.outlets) {
                                                    if (outlet.latitude != null && outlet.longitude != null)
                                                        subject.onNext(new QuotationMarker(outlet, quotation.price));
                                                }
                                            },
                                            error -> Log.fatalError(new RuntimeException(error)));
                        },
                        error -> Log.fatalError(new RuntimeException(error)));

            })
            .doOnError(throwable -> Log.fatalError(new RuntimeException(
                    "error response in subscribe after doOnSubscribe",
                    throwable)))
                    // combine with another observable that emits items regularly (every 100ms)
                    // so that a new event is received every 100ms :
                    // also, first event itself is delayed.
            .zipWith(Observable.interval(150, TimeUnit.MILLISECONDS),
                    (seller, aLong) -> seller)
            .onBackpressureBlock() // prevent zipWith Observer.interval from throwing MissingBackpressureException s
            .doOnError(throwable -> Log.fatalError(new RuntimeException(
                    "error response after onBackpressureBlock()",
                    throwable))); // <-- error is thrown here

跟踪:

    05-06 00:38:25.532  28106-28166/com.instano.buyer W/System.err? java.lang.RuntimeException: error response after onBackpressureBlock()
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at com.instano.retailer.instano.application.controller.Quotations.lambda$fetchQuotationMarkersForProduct$59(Quotations.java:67)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at com.instano.retailer.instano.application.controller.Quotations.access$lambda$5(Quotations.java)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at com.instano.retailer.instano.application.controller.Quotations$$Lambda$8.call(Unknown Source)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.Observable$11.onError(Observable.java:4193)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:65)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OperatorOnBackpressureBlock$BlockingSubscriber.complete(OperatorOnBackpressureBlock.java:81)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.util.BackpressureDrainManager.drain(BackpressureDrainManager.java:190)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.util.BackpressureDrainManager.terminateAndDrain(BackpressureDrainManager.java:129)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OperatorOnBackpressureBlock$BlockingSubscriber.onError(OperatorOnBackpressureBlock.java:68)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onError(OperatorZip.java:324)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:332)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OnSubscribeTimerPeriodically$1.call(OnSubscribeTimerPeriodically.java:51)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.Scheduler$Worker$1.call(Scheduler.java:120)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:390)
    05-06 00:38:25.572  28106-28166/com.instano.buyer W/System.err? at java.util.concurrent.FutureTask.run(FutureTask.java:234)
    05-06 00:38:25.582  28106-28166/com.instano.buyer W/System.err? at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:153)
    05-06 00:38:25.592  28106-28166/com.instano.buyer W/System.err? at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:267)
    05-06 00:38:25.602  28106-28166/com.instano.buyer W/System.err? at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1080)
    05-06 00:38:25.602  28106-28166/com.instano.buyer W/System.err? at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:573)
    05-06 00:38:25.602  28106-28166/com.instano.buyer W/System.err? at java.lang.Thread.run(Thread.java:841)
    05-06 00:38:25.602  28106-28166/com.instano.buyer W/System.err? Caused by: rx.exceptions.MissingBackpressureException
    05-06 00:38:25.612  28106-28166/com.instano.buyer W/System.err? at rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:349)
    05-06 00:38:25.642  28106-28166/com.instano.buyer W/System.err? at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:330)
    05-06 00:38:25.642  28106-28166/com.instano.buyer W/System.err? ... 10 more

PS:Log.fatalError(err)只是我的包装Android.util.Log.e(...)

编辑

经过大量的反复试验,这wont fix对我来说变得一团糟.zipWith(Observable.interval...)似乎是罪魁祸首和可能的框架错误.删除这些行(以及我的周期性发射功能)我的代码可以工作.我正在使用一个可能onNext从不同线程调用的主题,然后我在它上面执行Obeservable运算符.

Answer 1

我认为（但我不确定）问题在于您的背压配置位于zip操作员之后。

操作zip员需要缓冲一个项目Observable以将其与另一个项目一起压缩Observable。这是应该抛出异常的缓冲区。（看这里）

Observable为了解决您的问题，我认为您应该尝试在操作员中使用的一个（或每个）上添加反压配置zip。

例子 ：

obs.zipWith(Observable.interval(150, TimeUnit.MILLISECONDS).onBackPressureDrop());

obs.onBackPressureBlock().zipWith(Observable.interval(150, TimeUnit.MILLISECONDS));