RxBus、EventBus因为解耦太彻底，滥用的话，项目可维护性会越来越低；一些简单场景更推荐用回调、Subject来代替事件总线。
实际使用场景，如果RxBus，EventBus二选一，我更倾向于使用EventBus， RxJava专注工作流，EventBus专注事件总线，职责更清晰
几个月前，我写过一篇实现简单的RxBus文章： 用RxJava实现事件总线。
在实际环境中，你会发现RxBus还是有一些问题的。
- 你需要RxBus支持Sticky功能。
- 你会发现在你订阅了某个事件后，在后续接收到该事件时，处理的过程中发生了异常，你可能会发现后续的事件都接收不到了！
我将分2篇文章分别给出其方案，这篇介绍RxBus中的异常处理方案，另外一篇介绍如何实现Sticky:
深入RxBus：［支持Sticky事件］
异常处理
在使用RxBus过程中，你会发现你订阅了某个事件后，在后续接收到该事件时，如果处理的过程中发生了异常，你会发现后续的事件再也接收不到了，除非你重新订阅！
原因在于RxJava的事件序列机制，一个订阅事件是以onCompleted()或者onError()作为结束的，即：一旦订阅者的onCompleted()或onError()被调用，订阅者和被订阅者的订阅关系就解除了。
这里说下RxJava的异常传递机制：onError()在Observable序列传递过程中出现任何异常时被调用，然后终止Observable事件序列的传递，以此通知所有的订阅者发生了一个不可恢复的错误，即：异常总会传递到订阅者。
这本是RxJava的一个优点，反而在事件总线的场景下，成了让人头疼的问题！
所以我们的RxBus的订阅者在处理订阅事件时，一旦发生了异常，而又没Catch，那么最终都会调用到onError()，而一旦走到onError()，就意味着这个订阅者和该Subject解除了订阅关系，因此再也收不到后续发出的事件了～ 囧
我目前想到2种方案，如果你有更好的，欢迎告知我。
解决方案：自动重新订阅
即在onError(e)或onCompleted()发生时，立即重新订阅，保证订阅事件在解决时可以立即恢复。
private void subscribeEvent(){
RxBus.getDefault().toObservable(Event.class)
// 使用操作符过程中，无需try,catch，直接使用
.subscribe(new Subscriber<Event>() {
@Override
public void onCompleted() {
subscribeEvent();
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
subscribeEvent();
}
@Override
public void onNext(Event event) {
// 直接处理接收的事件
}
});
}
注意：这个方案如果用于Sticky事件，onError中重新订阅时，要使用RxBus.toObservable()而不是toObservableSticky()
原因在于Sticky的粘性特性，使用toObservableSticky()，引起error的事件如果重新订阅的话，该事件很可能继续导致error，从而引起死循环！
解决方案：onErrorResumeNext
该方案由简友lihansey提供，使用onErrorResumeNext()来catch链中发生的异常：
RxBus.getDefault().toObservableSticky(EventSticky.class) // 建议在Sticky时,在操作符内主动try,catch
.flatMap(event -> {
return Observable.juset(event)
.map(...) // 在flatMap里变换Observable
// 由于下面onErrorResumeNext, 因此 error 事件无法传递到observer, 故需要在这里做处理
.doOnError(e -> // todo)
.onErrorResumeNext(Observable.never())
})
.subscribe(new Action1<EventSticky>() {
@Override
public void call(EventSticky eventSticky) {
try {
// 处理接收的事件
} catch (Exception e) {
e.printStackTrace();
}
}
});
需要注意的是，变换操作需要在flatMap里的Observable上执行。
onErrorResumeNext会在Observable Error时开始发射一个新的Observable，从而让"catch"了flatMap里的Observable，不会执行到Obsever的onX方法，从而不会中断订阅链。
最后
附上一个Demo：
- 提供了使用Sticky特性的示例
- 异常处理的示例：让其在发生异常后，仍能正确接收到后续的Event。
Demo
参考：
ReactiveX: http://reactivex.io/
如果你有更好的解决方案，欢迎告知我，谢谢！
网友评论
个人觉得应该可以，只是这时重新订阅的时候使用
RxBus.getDefault().toObservable(EventSticky.class)，
而非RxBus.getDefault().toObservableSticky(EventSticky.class) ，这样就可以避免error的事件被重新发送，造成死循环。
2. onErrorResumeNext的解决方案很妙，在RxBus的使用场景下是没问题的，但这种处理error的策略不具通用性，稍换一种场景就不行了，比如想下面的代码，如果异常出现在原Observable中，subscriber的onError()方法依旧会被调用，订阅依旧会被终止：
Observable.create(new ObservableOnSubscribe<EventSticky>() {
@Override
public void subscribe(@NonNull ObservableEmitter<EventSticky> e) throws Exception {
int k = 1;
for (int j = 0; j < 10; j++) {
if (j == 5) {
e.onError(new NullPointerException("The " + k + "th element is null"));
}
e.onNext(new EventSticky("name: " + (k++)));
}
}
})
.flatMap(event -> {
return Observable.just(event)
.map(new Function<EventSticky, EventSticky>() {
@Override
public EventSticky apply(@NonNull EventSticky eventSticky) throws Exception {
throw new RuntimeException("fake error");
// return eventSticky;
}
}) // 在flatMap里变换Observable
// 由于下面onErrorResumeNext, 因此 error 事件无法传递到observer, 故需要在这里做处理
.doOnError(e -> {
System.err.println(e.getMessage());
})
.onErrorResumeNext(Observable.never());
})
.subscribe(eventSticky -> System.out.println(eventSticky.getName()),
throwable -> System.err.println("test:" + throwable.getMessage()));
代码得执行结果如下：
fake error
fake error
fake error
fake error
fake error
test:The 6th element is null
List<Subject> subjectList = subjectMapper.get(tag);
if (null != subjectList) {
for (final Subject subject : subjectList) {
subject.onNext(content);
}
}
"onNext"报错:
rx.exceptions.OnErrorNotImplementedException
rx.exceptions.MissingBackpressureException
1. 发送端：发送端主动发送异常，subscriber.onError(e)
2. 接收端：接收端处理出现了异常，onError(e)被调用
retry针对主动发送异常而言，所以retry不可以。
要阻止事件的"出错"or"正常结束", 应该像下面这么搞:
RxBus.getDefault().toObservableSticky(EventSticky.class)
.map(new Func1<EventSticky, EventSticky>() {
@Override
public EventSticky call(EventSticky eventSticky) {
throw new Exception("模拟出错!!!!");
return eventSticky;
}
})
.onErrorResumeNext(Observable.never()) // <--------- 把"错误" 转化成 Observable.never(), 记住是 never
.subscribe(new Action1<EventSticky>() {
@Override
public void call(EventSticky eventSticky) {
try {
// 处理接收的事件
} catch (Exception e) {
e.printStackTrace();
}
}
});