RxJava："java.lang.IllegalStateException: Only one subscriber allowed!"
RxJava: "java.lang.IllegalStateException: Only one subscriber allowed!"
我正在使用 RxJava 计算 Android 中某些传感器数据的归一化自相关。奇怪的是，我的代码抛出一个异常 ("java.lang.IllegalStateException: Only one subscriber allowed!")，我不确定该怎么做：我知道 GroupedObservables 在订阅我的多个订阅者时可能会抛出这个异常，但我不认为我是在任何地方使用这样的东西。
您在下面找到（最有可能）触发异常的方法：
public Observable<Float> normalizedAutoCorrelation(Observable<Float> observable, final int lag) {
Observable<Float> laggedObservable = observable.skip(lag);
Observable<Float> meanObservable = mean(observable, lag);
Observable<Float> laggedMeanObservable = mean(laggedObservable, lag);
Observable<Float> standardDeviationObservable = standardDeviation(observable, meanObservable, lag);
Observable<Float> laggedStandardDeviationObservable = standardDeviation(laggedObservable, laggedMeanObservable, lag);
Observable<Float> deviation = observable.zipWith(meanObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float value, Float mean) {
return value - mean;
}
});
Observable<Float> laggedDeviation = observable.zipWith(laggedMeanObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float value, Float mean) {
return value - mean;
}
});
Observable<Float> autoCorrelationPartObservable = deviation.zipWith(laggedDeviation, new Func2<Float, Float, Float>() {
@Override
public Float call(Float value, Float laggedValue) {
return value * laggedValue;
}
});
Observable<Float> autoCorrelationObservable = flatten(autoCorrelationPartObservable.window(lag, 1).scan(new Func2<Observable<Float>, Observable<Float>, Observable<Float>>() {
@Override
public Observable<Float> call(Observable<Float> memoObservable, Observable<Float> observable) {
if(memoObservable == null) return observable;
return memoObservable.zipWith(observable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float memo, Float value) {
return memo + value;
}
});
}
}));
Observable<Float> normalizationObservable = standardDeviationObservable.zipWith(laggedStandardDeviationObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float standardDeviation, Float laggedStandardDeviation) {
return lag * standardDeviation * laggedStandardDeviation;
}
});
return autoCorrelationObservable.zipWith(normalizationObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float autoCorrelation, Float normalization) {
return autoCorrelation / normalization;
}
});
}
这是我得到的堆栈跟踪：
java.lang.IllegalStateException: Only one subscriber allowed!
at rx.internal.operators.BufferUntilSubscriber$OnSubscribeAction.call(BufferUntilSubscriber.java:124)
at rx.internal.operators.BufferUntilSubscriber$OnSubscribeAction.call(BufferUntilSubscriber.java:81)
at rx.Observable.unsafeSubscribe(Observable.java:7304)
at rx.internal.operators.OperatorZip$Zip.start(OperatorZip.java:210)
at rx.internal.operators.OperatorZip$ZipSubscriber.onNext(OperatorZip.java:154)
at rx.internal.operators.OperatorZip$ZipSubscriber.onNext(OperatorZip.java:120)
at rx.internal.util.ScalarSynchronousObservable.call(ScalarSynchronousObservable.java:41)
at rx.internal.util.ScalarSynchronousObservable.call(ScalarSynchronousObservable.java:30)
at rx.Observable.call(Observable.java:145)
at rx.Observable.call(Observable.java:137)
at rx.Observable.unsafeSubscribe(Observable.java:7304)
at rx.internal.operators.OperatorMerge$MergeSubscriber.handleNewSource(OperatorMerge.java:188)
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:158)
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:93)
at rx.internal.operators.OperatorMap.onNext(OperatorMap.java:55)
at rx.internal.operators.OperatorScan.onNext(OperatorScan.java:110)
at rx.internal.operators.OperatorWindowWithSize$InexactSubscriber.onNext(OperatorWindowWithSize.java:173)
at rx.internal.operators.OperatorZip$Zip.tick(OperatorZip.java:255)
at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:326)
at rx.internal.operators.OperatorZip$Zip.tick(OperatorZip.java:255)
at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:326)
at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:635)
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:545)
at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
at rx.internal.operators.TakeLastQueueProducer.emit(TakeLastQueueProducer.java:98)
at rx.internal.operators.TakeLastQueueProducer.startEmitting(TakeLastQueueProducer.java:45)
at rx.internal.operators.OperatorTakeLast.onCompleted(OperatorTakeLast.java:59)
at rx.internal.operators.OperatorScan.onCompleted(OperatorScan.java:121)
at rx.internal.operators.BufferUntilSubscriber.onCompleted(BufferUntilSubscriber.java:161)
at rx.internal.operators.OperatorWindowWithSize$InexactSubscriber.onNext(OperatorWindowWithSize.java:183)
at rx.internal.operators.OperatorSkip.onNext(OperatorSkip.java:58)
at rx.internal.operators.OperatorMap.onNext(OperatorMap.java:55)
at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onNext(SubjectSubscriptionManager.java:224)
at rx.subjects.PublishSubject.onNext(PublishSubject.java:121)
at com.github.joopaue.smartphonesensing.SensorService.onSensorChanged(SensorService.java:102)
at android.hardware.SystemSensorManager$SensorEventQueue.dispatchSensorEvent(SystemSensorManager.java:418)
at android.os.MessageQueue.nativePollOnce(Native Method)
at android.os.MessageQueue.next(MessageQueue.java:138)
at android.os.Looper.loop(Looper.java:123)
at android.app.ActivityThread.main(ActivityThread.java:5146)
at java.lang.reflect.Method.invokeNative(Native Method)
at java.lang.reflect.Method.invoke(Method.java:515)
at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:732)
at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:566)
at dalvik.system.NativeStart.main(Native Method)
我不认为我在这里做了什么奇怪的事情：一些压缩、缩小、扫描和平面图。
我是不是漏掉了一些非常明显的东西，是不是我在这里违反了一些隐藏的规则，或者这是 RxJava 中的一个错误？谢谢！
PS。如果缺少一些代码让您无法得出结论，请直接提问，我会 post!
在 RxJava 中，操作符 groupBy 和 window return 一个只能被订阅一次的可观察对象，如果被订阅，它们会向唯一的订阅者重放他们累积的内容并切换到 'hot' 模式。
这是在 return 完全热的可观察对象和风险缺失值或 return 允许任何订阅者但无限期保留累积内容的无限制重放可观察对象之间的权衡。
中间地带，即单个订阅者，先冷后热的可观察性被认为是最不令人惊讶的行为，它使开发人员可以选择应用更多运算符并选择两个极端之间的任何一点：
source.window(1, TimeUnit.SECONDS)
.map(w -> w.publish())
.doOnNext(w -> w.connect())
.subscribe(...)
source.window(1, TimeUnit.SECONDS)
.map(w -> w.cache())
.subscribe(...)
我正在使用 RxJava 计算 Android 中某些传感器数据的归一化自相关。奇怪的是，我的代码抛出一个异常 ("java.lang.IllegalStateException: Only one subscriber allowed!")，我不确定该怎么做：我知道 GroupedObservables 在订阅我的多个订阅者时可能会抛出这个异常，但我不认为我是在任何地方使用这样的东西。
您在下面找到（最有可能）触发异常的方法：
public Observable<Float> normalizedAutoCorrelation(Observable<Float> observable, final int lag) {
Observable<Float> laggedObservable = observable.skip(lag);
Observable<Float> meanObservable = mean(observable, lag);
Observable<Float> laggedMeanObservable = mean(laggedObservable, lag);
Observable<Float> standardDeviationObservable = standardDeviation(observable, meanObservable, lag);
Observable<Float> laggedStandardDeviationObservable = standardDeviation(laggedObservable, laggedMeanObservable, lag);
Observable<Float> deviation = observable.zipWith(meanObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float value, Float mean) {
return value - mean;
}
});
Observable<Float> laggedDeviation = observable.zipWith(laggedMeanObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float value, Float mean) {
return value - mean;
}
});
Observable<Float> autoCorrelationPartObservable = deviation.zipWith(laggedDeviation, new Func2<Float, Float, Float>() {
@Override
public Float call(Float value, Float laggedValue) {
return value * laggedValue;
}
});
Observable<Float> autoCorrelationObservable = flatten(autoCorrelationPartObservable.window(lag, 1).scan(new Func2<Observable<Float>, Observable<Float>, Observable<Float>>() {
@Override
public Observable<Float> call(Observable<Float> memoObservable, Observable<Float> observable) {
if(memoObservable == null) return observable;
return memoObservable.zipWith(observable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float memo, Float value) {
return memo + value;
}
});
}
}));
Observable<Float> normalizationObservable = standardDeviationObservable.zipWith(laggedStandardDeviationObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float standardDeviation, Float laggedStandardDeviation) {
return lag * standardDeviation * laggedStandardDeviation;
}
});
return autoCorrelationObservable.zipWith(normalizationObservable, new Func2<Float, Float, Float>() {
@Override
public Float call(Float autoCorrelation, Float normalization) {
return autoCorrelation / normalization;
}
});
}
这是我得到的堆栈跟踪：
java.lang.IllegalStateException: Only one subscriber allowed!
at rx.internal.operators.BufferUntilSubscriber$OnSubscribeAction.call(BufferUntilSubscriber.java:124)
at rx.internal.operators.BufferUntilSubscriber$OnSubscribeAction.call(BufferUntilSubscriber.java:81)
at rx.Observable.unsafeSubscribe(Observable.java:7304)
at rx.internal.operators.OperatorZip$Zip.start(OperatorZip.java:210)
at rx.internal.operators.OperatorZip$ZipSubscriber.onNext(OperatorZip.java:154)
at rx.internal.operators.OperatorZip$ZipSubscriber.onNext(OperatorZip.java:120)
at rx.internal.util.ScalarSynchronousObservable.call(ScalarSynchronousObservable.java:41)
at rx.internal.util.ScalarSynchronousObservable.call(ScalarSynchronousObservable.java:30)
at rx.Observable.call(Observable.java:145)
at rx.Observable.call(Observable.java:137)
at rx.Observable.unsafeSubscribe(Observable.java:7304)
at rx.internal.operators.OperatorMerge$MergeSubscriber.handleNewSource(OperatorMerge.java:188)
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:158)
at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:93)
at rx.internal.operators.OperatorMap.onNext(OperatorMap.java:55)
at rx.internal.operators.OperatorScan.onNext(OperatorScan.java:110)
at rx.internal.operators.OperatorWindowWithSize$InexactSubscriber.onNext(OperatorWindowWithSize.java:173)
at rx.internal.operators.OperatorZip$Zip.tick(OperatorZip.java:255)
at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:326)
at rx.internal.operators.OperatorZip$Zip.tick(OperatorZip.java:255)
at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:326)
at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:635)
at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:545)
at rx.internal.operators.NotificationLite.accept(NotificationLite.java:150)
at rx.internal.operators.TakeLastQueueProducer.emit(TakeLastQueueProducer.java:98)
at rx.internal.operators.TakeLastQueueProducer.startEmitting(TakeLastQueueProducer.java:45)
at rx.internal.operators.OperatorTakeLast.onCompleted(OperatorTakeLast.java:59)
at rx.internal.operators.OperatorScan.onCompleted(OperatorScan.java:121)
at rx.internal.operators.BufferUntilSubscriber.onCompleted(BufferUntilSubscriber.java:161)
at rx.internal.operators.OperatorWindowWithSize$InexactSubscriber.onNext(OperatorWindowWithSize.java:183)
at rx.internal.operators.OperatorSkip.onNext(OperatorSkip.java:58)
at rx.internal.operators.OperatorMap.onNext(OperatorMap.java:55)
at rx.subjects.SubjectSubscriptionManager$SubjectObserver.onNext(SubjectSubscriptionManager.java:224)
at rx.subjects.PublishSubject.onNext(PublishSubject.java:121)
at com.github.joopaue.smartphonesensing.SensorService.onSensorChanged(SensorService.java:102)
at android.hardware.SystemSensorManager$SensorEventQueue.dispatchSensorEvent(SystemSensorManager.java:418)
at android.os.MessageQueue.nativePollOnce(Native Method)
at android.os.MessageQueue.next(MessageQueue.java:138)
at android.os.Looper.loop(Looper.java:123)
at android.app.ActivityThread.main(ActivityThread.java:5146)
at java.lang.reflect.Method.invokeNative(Native Method)
at java.lang.reflect.Method.invoke(Method.java:515)
at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:732)
at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:566)
at dalvik.system.NativeStart.main(Native Method)
我不认为我在这里做了什么奇怪的事情：一些压缩、缩小、扫描和平面图。
我是不是漏掉了一些非常明显的东西，是不是我在这里违反了一些隐藏的规则，或者这是 RxJava 中的一个错误？谢谢！
PS。如果缺少一些代码让您无法得出结论，请直接提问，我会 post!
在 RxJava 中，操作符 groupBy 和 window return 一个只能被订阅一次的可观察对象，如果被订阅，它们会向唯一的订阅者重放他们累积的内容并切换到 'hot' 模式。
这是在 return 完全热的可观察对象和风险缺失值或 return 允许任何订阅者但无限期保留累积内容的无限制重放可观察对象之间的权衡。
中间地带，即单个订阅者，先冷后热的可观察性被认为是最不令人惊讶的行为，它使开发人员可以选择应用更多运算符并选择两个极端之间的任何一点：
source.window(1, TimeUnit.SECONDS)
.map(w -> w.publish())
.doOnNext(w -> w.connect())
.subscribe(...)
source.window(1, TimeUnit.SECONDS)
.map(w -> w.cache())
.subscribe(...)