我用rxbus連續發送大量消息報錯

08-26 09:37:13.458 10637-10637/com.dituwuyou E/AndroidRuntime: FATAL EXCEPTION: main

Process: com.dituwuyou, PID: 10637java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:112)at android.os.Handler.handleCallback(Handler.java:739)at android.os.Handler.dispatchMessage(Handler.java:95)at android.os.Looper.loop(Looper.java:148)at android.app.ActivityThread.main(ActivityThread.java:5417)at java.lang.reflect.Method.invoke(Native Method)at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726)at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616) Caused by: rx.exceptions.OnErrorNotImplementedException: PublishSubject: could not emit value due to lack of requestsat rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:386)at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:383)at rx.internal.util.ActionSubscriber.onError(ActionSubscriber.java:44)at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:152)at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:115)at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:276)at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:219)at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:107)at android.os.Handler.handleCallback(Handler.java:739)?at a<愛尬聊_百科知識>ndroid.os.Handler.dispatchMessage(Handler.java:95)?at android.os.Looper.loop(Looper.java:148)?at android.app.ActivityThread.main(ActivityThread.java:5417)?at java.lang.reflect.Method.invoke(Native Method)?at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726)?at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616)? Caused by: rx.exceptions.MissingBackpressureException: PublishSubject: could not emit value due to lack of requestsat rx.subjects.PublishSubject$PublishSubjectProducer.onNext(PublishSubject.java:308)at rx.subjects.PublishSubject$PublishSubjectState.onNext(PublishSubject.java:220)at rx.subjects.PublishSubject.onNext(PublishSubject.java:73)at rx.observers.SerializedObserver.onNext(SerializedObserver.java:92)at rx.subjects.SerializedSubject.onNext(SerializedSubject.java:67)at com.dituwuyou.widget.rxjava.RxBus.send(RxBus.java:45)at com.dituwuyou.joint.CoorSocketService.messageReceived(CoorSocketService.java:51)at com.dituwuyou.fayeclient.FayeClient.parseFayeMessage(FayeClient.java:535)at com.dituwuyou.fayeclient.FayeClient.onMessage(FayeClient.java:390)at com.dituwuyou.fayeclient.HybiParser.emitFrame(HybiParser.java:304)at com.dituwuyou.fayeclient.HybiParser.start(HybiParser.java:130)at com.dituwuyou.fayeclient.WebSocketClient$1.run(WebSocketClient.java:119)at java.lang.Thread.run(Thread.java:818)

我的rxbus是這樣定義的

import rx.Observable; import rx.subjects.PublishSubject; import rx.subjects.SerializedSubject; import rx.subjects.Subject;/*** Created by xg on 2016/6/24.* 消息傳遞(替換handler，eventbus)*/ public class RxBus { private static volatile RxBus mInstance; private final Subject bus;public RxBus() { bus = new SerializedSubject<>(PublishSubject.create()); }/*** 單例模式RxBus** @return*/ public static RxBus getRxBusSingleton() { RxBus rxBus2 = mInstance; if (mInstance == null) { synchronized (RxBus.class) { rxBus2 = mInstance; if (mInstance == null) { rxBus2 = new RxBus(); mInstance = rxBus2; } } } return rxBus2; }/*** 發送消息** @param object*/ public void send(Object object) { bus.onNext(object); }/*** 接收消息** @return*/ public Observable toObserverable() { return bus; } }

這是第45行
bus.onNext(object);
應該是出現背壓了

---

弘開#p#何老三 2022-07-13 14:39

1.看不懂你為什么要加個rxBus22.要處理onError的情況3.先注冊 后send

---

凌晚軒#p#自尊自愛 2022-07-13 14:53

用的什么版本的RxJava. 在1.1.6版本中已經沒有rx.subjects.PublishSubject$PublishSubjectProducer這個類了

---