RxBus并不是一个库，是利用RxJava的特性构建的一种数据传递的工具模式。相信大多数开发者都使用过EventBus，作为事件总线通信库，如果你的项目已经加入RxJava和EventBus，不妨用RxBus代替EventBus，以减少库的依赖。

与广播相比，RxBus有以下优势：

1.可以处处发送，处处接收，没有广播那种多处发送，仅在onReceiver()接收处理的限制。

2.异步处理，可以执行大数据量的操作，没有onReceiver处理性能的限制。

一、添加RxJava和RxAndroid依赖

//RxJava and RxAndroid compile 'io.reactivex:rxandroid:1.1.0' compile 'io.reactivex:rxjava:1.1.0'

二、新建RxBus类

不多说直接上代码：

import rx.Observable; import rx.subjects.PublishSubject; import rx.subjects.SerializedSubject; import rx.subjects.Subject; /** * Created by xialo on 2016/6/28. */ public class RxBus { private static volatile RxBus mInstance; private final Subject bus; public RxBus() { bus = new SerializedSubject<>(PublishSubject.create()); } /** * 单例模式RxBus * * @return */ public static RxBus getInstance() { 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 post(Object object) { bus.onNext(object); } /** * 接收消息 * * @param eventType * @param <T> * @return */ public <T> Observable<T> toObserverable(Class<T> eventType) { return bus.ofType(eventType); } }

1、Subject同时充当了Observer和Observable的角色，Subject是非线程安全的，要避免该问题，需要将 Subject转换为一个 SerializedSubject，上述RxBus类中把线程非安全的PublishSubject包装成线程安全的Subject。 2、PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。 3、ofType操作符只发射指定类型的数据，其内部就是filter+cast

三、创建你需要发送的事件类

我们这里用StudentEvent举例

/** * Created by xialo on 2016/6/28. */ public class StudentEvent { private String id; private String name; public StudentEvent(String id, String name) { this.id = id; this.name = name; } public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } }

四、发送事件

RxBus.getInstance().post(new StudentEvent("007","小明"));

五、接收事件

rxSbscription=RxBus.getInstance().toObserverable(StudentEvent.class) .subscribe(new Action1<StudentEvent>() { @Override public void call(StudentEvent studentEvent) { textView.setText("id:"+ studentEvent.getId()+" name:"+ studentEvent.getName()); } });

注：rxSbscription是Sbscription的对象，我们这里把RxBus.getInstance().toObserverable(StudentEvent.class)赋值给rxSbscription以方便生命周期结束时取消订阅事件

六、取消订阅

@Override protected void onDestroy() { if (!rxSbscription.isUnsubscribed()){ rxSbscription.unsubscribe(); } super.onDestroy(); }

七、UI线程切换

Rxbus默认在子线程异步处理数据，如果你希望在接收到数据后立刻对UI进行相关处理，更改UI界面，那么将会报错。你可以利用handler线程转发数据进行UI操作，或者可以转换为UI(主)线程中处理，如下切换为主线程操作，这样你就可以在handleBroadcast(object.getType(),object.getValue())中直接操作value了，如将字符串value让TextView显示。

      rxSbscription= RxBus.getDefault().toObservable(JniObject.class).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<JniObject>() { //                               @Override //                               public void call(JniObject object) { //                                   handleBroadcast(object.getType(),object.getValue()); //                               } //                           }, //                        new Action1<Throwable>() { //                            @Override //                            public void call(Throwable throwable) { //                                // TODO: 处理异常 //                            } //                        }); //    }

参考： http://wuxiaolong.me/2016/04/07/rxbus/ http://www.jianshu.com/p/ca090f6e2fe2