文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >问答首页 >对我的操作员？

问对我的操作员？
EN

Stack Overflow用户
提问于 2016-02-09 20:49:47
回答 1查看 275关注 0票数 2

我实现了我自己的名为DoCountOperator的客户操作符，它支持我创建的自定义Observable操作符，如doOnNextCount()、doOnErrorCount()和doOnCompletedCount()计数。它允许为这些事件创建一个副作用，并在这些事件中使用排放计数来做一些事情。例如，doOnCompletedCount()允许您在onCompleted()事件之后创建完全发射计数的副作用。

但是，我遇到了大量排放的MissingBackpressureException问题，我的堆栈跟踪提到了我的操作员。这是Operator..。

代码语言：javascript
复制
final class DoCountOperator<T> implements Observable.Operator<T,T> {

    interface CountObserver {
        void onNext(int emissionCount);
        void onError(int emissionCount);
        void onCompleted(int emissionCount);
    }

    private final DoCountOperator.CountObserver doObserver;

    DoCountOperator(DoCountOperator.CountObserver doObserver) {
        this.doObserver = doObserver;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
        return new Subscriber<T>() {
            private int count = 0;
            private boolean done = false;

            @Override
            public void onCompleted() {
                if (done) {
                    return;
                }
                try {
                    doObserver.onCompleted(count);
                }catch (Throwable throwable) {
                    //Exceptions.throwIfFatal(throwable);
                    onError(throwable);
                }

                done = true;
                child.onCompleted();
            }

            @Override
            public void onError(Throwable throwable) {
                if (done) {
                    return;
                }
                try {
                    doObserver.onError(count);
                } catch (Throwable throwable1) {
                    throwable1.printStackTrace();
                }
                child.onError(throwable);
            }

            @Override
            public void onNext(T t) {
                if (done) {
                    return;
                }
                try {
                    doObserver.onNext(++count);
                }catch (Throwable throwable) {
                    Exceptions.throwIfFatal(throwable);
                }
                child.onNext(t);
            }
        };
    }
}

下面是doOnCompletedCount()如何使用它。当我发出大约50,000到90,000个发出的项目时，我会得到一个错误。

代码语言：javascript
复制
public static <T> Observable.Operator<T,T> doOnCompletedCount(final IntConsumer countAction) {
        return new DoCountOperator<>(new DoCountOperator.CountObserver() {
            @Override
            public void onNext(int emissionCount) {

            }

            @Override
            public void onError(int emissionCount) {

            }

            @Override
            public void onCompleted(int emissionCount) {
                countAction.accept(emissionCount);
            }
        });
    }

我想不出这个操作者怎么会被背压压垮，但也许它没有把背压问题传达到整个链条上？我到底做错了什么？这是堆栈追踪..。

代码语言：javascript
复制
rx.exceptions.MissingBackpressureException
                at rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:352)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.queueScalar(OperatorMerge.java:346)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:329)
                at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:804)
                at rx.internal.operators.OperatorScan$2.onNext(OperatorScan.java:117)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at com.swa.rm.common.rx.DoCountOperator$1.onNext(DoCountOperator.java:66)
                at rx.internal.operators.OperatorFinally$1.onNext(OperatorFinally.java:48)
                at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:85)
                at rx.observers.SerializedObserver.onNext(SerializedObserver.java:95)
                at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95)
                at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:199)
                at com.swa.rm.common.rx.RxOperators$1.onNext(RxOperators.java:49)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OperatorFinally$1.onNext(OperatorFinally.java:48)
                at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:85)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:365)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:327)
                at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:804)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:365)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:327)
                at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:804)
                at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
                at rx.internal.operators.CachedObservable$ReplayProducer.replay(CachedObservable.java:404)
                at rx.internal.operators.CachedObservable$CacheState.dispatch(CachedObservable.java:220)
                at rx.internal.operators.CachedObservable$CacheState.onNext(CachedObservable.java:191)
                at rx.internal.operators.CachedObservable$CacheState$1.onNext(CachedObservable.java:171)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:365)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:327)
                at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:804)
                at rx.observers.SerializedObserver.onNext(SerializedObserver.java:95)
                at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95)
                at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:199)
                at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:85)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OperatorTake$1.onNext(OperatorTake.java:73)
                at rx.observers.SerializedObserver.onNext(SerializedObserver.java:95)
                at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95)
                at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:199)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.observers.SerializedObserver.onNext(SerializedObserver.java:95)
                at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95)
                at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:199)
                at rx.internal.operators.OperatorFilter$1.onNext(OperatorFilter.java:54)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:46)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:35)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:172)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:136)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:79)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:46)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:35)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:172)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:136)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:79)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.observers.SerializedObserver.onNext(SerializedObserver.java:95)
                at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95)
                at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:199)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.observers.SerializedObserver.onNext(SerializedObserver.java:95)
                at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:95)
                at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onNext(OperatorConcat.java:199)
                at rx.internal.operators.OperatorFilter$1.onNext(OperatorFilter.java:54)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:46)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:35)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:172)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:136)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:79)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:46)
                at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:35)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:172)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:136)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:79)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OperatorFilter$1.onNext(OperatorFilter.java:54)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:365)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:327)
                at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:804)
                at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.slowpath(OnSubscribeFromIterable.java:97)
                at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.request(OnSubscribeFromIterable.java:73)
                at rx.Subscriber.setProducer(Subscriber.java:211)
                at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:49)
                at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:32)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:232)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:142)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
                at rx.internal.operators.CachedObservable$ReplayProducer.replay(CachedObservable.java:404)
                at rx.internal.operators.CachedObservable$ReplayProducer.request(CachedObservable.java:304)
                at rx.Subscriber.setProducer(Subscriber.java:211)
                at rx.Subscriber.setProducer(Subscriber.java:205)
                at rx.Subscriber.setProducer(Subscriber.java:205)
                at rx.Subscriber.setProducer(Subscriber.java:205)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:244)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:230)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:172)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:136)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onNext(OperatorConcat.java:79)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.slowpath(OnSubscribeFromIterable.java:97)
                at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.request(OnSubscribeFromIterable.java:73)
                at rx.Subscriber.setProducer(Subscriber.java:211)
                at rx.Subscriber.setProducer(Subscriber.java:205)
                at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:49)
                at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:32)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:232)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:142)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
                at rx.internal.operators.CachedObservable$ReplayProducer.replay(CachedObservable.java:404)
                at rx.internal.operators.CachedObservable$CacheState.dispatch(CachedObservable.java:220)
                at rx.internal.operators.CachedObservable$CacheState.onNext(CachedObservable.java:191)
                at rx.internal.operators.CachedObservable$CacheState$1.onNext(CachedObservable.java:171)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:365)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:327)
                at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:804)
                at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54)
                at rx.internal.producers.SingleDelayedProducer.emit(SingleDelayedProducer.java:102)
                at rx.internal.producers.SingleDelayedProducer.setValue(SingleDelayedProducer.java:85)
                at rx.internal.operators.OperatorToObservableSortedList$2.onCompleted(OperatorToObservableSortedList.java:82)
                at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
                at rx.internal.operators.OperatorGroupBy$GroupBySubscriber$2$2.onCompleted(OperatorGroupBy.java:264)
                at rx.observers.Subscribers$5.onCompleted(Subscribers.java:220)
                at rx.internal.operators.BufferUntilSubscriber.onCompleted(BufferUntilSubscriber.java:155)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:140)
                at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.emitItem(OperatorGroupBy.java:339)
                at rx.internal.operators.OperatorGroupBy$GroupBySubscriber.onCompleted(OperatorGroupBy.java:161)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitLoop(OperatorMerge.java:614)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emit(OperatorMerge.java:526)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:255)
                at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:140)
                at rx.internal.operators.CachedObservable$ReplayProducer.replay(CachedObservable.java:404)
                at rx.internal.operators.CachedObservable$CacheState.dispatch(CachedObservable.java:220)
                at rx.internal.operators.CachedObservable$CacheState.onCompleted(CachedObservable.java:211)
                at rx.internal.operators.CachedObservable$CacheState$1.onCompleted(CachedObservable.java:179)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitLoop(OperatorMerge.java:614)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emit(OperatorMerge.java:526)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:255)
                at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emitLoop(OperatorMerge.java:614)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.emit(OperatorMerge.java:526)
                at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:255)
                at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:140)
                at rx.internal.operators.CachedObservable$ReplayProducer.replay(CachedObservable.java:404)
                at rx.internal.operators.CachedObservable$CacheState.dispatch(CachedObservable.java:220)
                at rx.internal.operators.CachedObservable$CacheState.onCompleted(CachedObservable.java:211)
                at rx.internal.operators.CachedObservable$CacheState$1.onCompleted(CachedObservable.java:179)
                at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:183)
                at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:65)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:167)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onCompleted(OperatorConcat.java:150)
                at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
                at rx.internal.operators.OperatorBufferWithSize$1.onCompleted(OperatorBufferWithSize.java:126)
                at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:183)
                at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:65)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:167)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onCompleted(OperatorConcat.java:150)
                at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
                at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:183)
                at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:65)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:167)
                at rx.internal.operators.OperatorConcat$ConcatSubscriber.onCompleted(OperatorConcat.java:150)
                at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.slowpath(OnSubscribeFromIterable.java:101)
                at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.request(OnSubscribeFromIterable.java:73)
                at rx.Subscriber.setProducer(Subscriber.java:211)
                at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:49)
                at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:32)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.CachedObservable$CacheState.connect(CachedObservable.java:183)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:248)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:230)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.CachedObservable$CacheState.connect(CachedObservable.java:183)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:248)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:230)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.CachedObservable$CacheState.connect(CachedObservable.java:183)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:248)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:230)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable$2.call(Observable.java:162)
                at rx.Observable$2.call(Observable.java:154)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.CachedObservable$CacheState.connect(CachedObservable.java:183)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:248)
                at rx.internal.operators.CachedObservable$CachedSubscribe.call(CachedObservable.java:230)
                at rx.Observable.unsafeSubscribe(Observable.java:8098)
                at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62)
                at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
                at rx.schedulers.ExecutorScheduler$ExecutorSchedulerWorker.run(ExecutorScheduler.java:98)
                at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
                at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
                at java.lang.Thread.run(Thread.java:745)
java
rx-java
reactive-programming
EN

回答 1

Stack Overflow用户

回答已采纳

发布于 2016-02-09 22:48:06

您忘记链接下游订阅者的取消订阅和背压“通道”在您的代码。你需要这样的东西：

代码语言：javascript
复制
final class DoCountOperator<T> implements Observable.Operator<T,T> {

    interface CountObserver {
        void onNext(int emissionCount);
        void onError(int emissionCount);
        void onCompleted(int emissionCount);
    }

    private final DoCountOperator.CountObserver doObserver;

    DoCountOperator(DoCountOperator.CountObserver doObserver) {
        this.doObserver = doObserver;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
        Subscriber<T> parent = new Subscriber<T>() {
            private int count = 0;
            private boolean done = false;

            @Override
            public void onCompleted() {
                if (done) {
                    return;
                }
                try {
                    doObserver.onCompleted(count);
                }catch (Throwable throwable) {
                    //Exceptions.throwIfFatal(throwable);
                    onError(throwable);
                }

                done = true;
                child.onCompleted();
            }

            @Override
            public void onError(Throwable throwable) {
                if (done) {
                    return;
                }
                try {
                    doObserver.onError(count);
                } catch (Throwable throwable1) {
                    throwable1.printStackTrace();
                }
                child.onError(throwable);
            }

            @Override
            public void onNext(T t) {
                if (done) {
                    return;
                }
                try {
                    doObserver.onNext(++count);
                }catch (Throwable throwable) {
                    Exceptions.throwIfFatal(throwable);
                }
                child.onNext(t);
            }

            @Override
            public void setProducer(Producer p) {
                child.setProducer(p);
            }
        };

        child.add(parent);

        return parent;
    }
}

因此，您不必返回自定义的订阅服务器on ()，而是将其放入一个局部变量中并调用child.add(parent)，从而建立取消订阅链。要使背压工作，最好的方法是覆盖自定义订阅服务器中的setProducer，并使用接收到的Producer实例调用child.setProducer。

票数 3
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：

https://stackoverflow.com/questions/35301902

复制
相关文章

相似问题

领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档