前面说过Rxjava的功能很强大，不仅仅是实现链式的异步操作，它的功能很强大还可以通过RxBus实现EventBus的消息/事件传递功
能，我们来看看
RxBus
1 package com.example.liujian.rxjavademo; 2 3 import java.util.HashMap; 4 5 import rx.Observable; 6 import rx.Subscription; 7 import rx.android.schedulers.AndroidSchedulers; 8 import rx.functions.Action0; 9 import rx.functions.Action1; 10 import rx.schedulers.Schedulers; 11 import rx.subjects.PublishSubject; 12 import rx.subjects.SerializedSubject; 13 import rx.subscriptions.CompositeSubscription; 14 15 /** 16 * @project_Name: RxjavaDemo 17 * @package: com.example.liujian.rxjavademo 18 * @description: 使用Rxjava,RxAndroid实现RxBus 19 * @author: liujian 20 * @date: 2016/10/5 11:08 21 * @version: V1.0 22 */ 23 24 public class RxBus { 25 private static volatile RxBus mInstance; 26 //Subject继承了Observable类又实现了Observer接口， Subject可以同时担当订阅者和被订阅者的角色 27 private SerializedSubject mSubject; 28 //一个类产生多个Subscription对象，用一CompositeSubscription 存储起来，以进行批量的取消订阅。避免内存泄漏 29 private HashMap mSubscriptionHashMap; 30 private RxBus(){ 31 //Subject是非线程安全的,SerializedSubject将PublishSubject 转换成一个线程安全的Subject对象 32 mSubject=new SerializedSubject<>(PublishSubject.create()); 33 } 34 public static RxBus getInstance(){ 35 if(mInstance==null){ 36 synchronized (RxBus.class){ 37 if(mInstance==null){ 38 mInstance=new RxBus(); 39 } 40 } 41 } 42 return mInstance; 43 } 44 /** 45 * 发生消息 46 */ 47 public void post(Object o){ 48 mSubject.onNext(o); 49 } 50 51 /** 52 * 返回指定类型的Observable实例 53 * @param type:要处理的消息的类型 54 * @param 55 * @return 56 */ 57 public Observable toObservable(final Class type){ 58 return mSubject.ofType(type); 59 } 60 61 /** 62 * 是否已含有观察者订阅 63 * @return 64 */ 65 public boolean hasObservers(){ 66 return mSubject.hasObservers(); 67 } 68 69 /** 70 * 默认的订阅方法 71 * @param 72 * @return 73 */ 74 public Subscription doSubscribe(Class type, Action1 next){ 75 return toObservable(type) 76 .subscribeOn(Schedulers.io()) 77 .observeOn(AndroidSchedulers.mainThread()) 78 .subscribe(next); 79 } 80 /** 81 * 默认的订阅方法 82 * @param 83 * @return 84 */ 85 public Subscription doSubscribe(Class type, Action1 next,Action1 error){ 86 return toObservable(type) 87 .subscribeOn(Schedulers.io()) 88 .observeOn(AndroidSchedulers.mainThread()) 89 .subscribe(next,error); 90 } 91 /** 92 * 默认的订阅方法 93 * @param 94 * @return 95 */ 96 public Subscription doSubscribe(Class type, Action1 next, Action1 error, Action0 complete){ 97 return toObservable(type) 98 .subscribeOn(Schedulers.io()) 99 .observeOn(AndroidSchedulers.mainThread())100 .subscribe(next,error,complete);101 }102 103 /**104 * 保存订阅后的subscription,方便一次性取消订阅105 * @param o106 * @param subscription107 */108 public void addSubscription(Object o,Subscription subscription){109 if(mSubscriptionHashMap==null){110 mSubscriptionHashMap=new HashMap<>();111 }112 String key=o.getClass().getSimpleName();113 if(mSubscriptionHashMap.containsKey(key)){114 mSubscriptionHashMap.get(key).add(subscription);115 }else{116 CompositeSubscription compositeSubscription=new CompositeSubscription();117 compositeSubscription.add(subscription);118 mSubscriptionHashMap.put(key,compositeSubscription);119 }120 }121 122 /**123 * 取消订阅124 * @param o125 */126 public void unSubscribe(Object o){127 if(mSubscriptionHashMap==null){128 return;129 }130 String key=o.getClass().getSimpleName();131 if(!mSubscriptionHashMap.containsKey(key)){132 return;133 }134 if(mSubscriptionHashMap.get(key)!=null){135 mSubscriptionHashMap.get(key).unsubscribe();136 }137 mSubscriptionHashMap.remove(key);138 }139 } 不多解释了，注释都写的很清楚了，我们来看一下如何使用
发送消息：
RxBus.getInstance().post("这是发送的消息"); 处理消息：
public void doSubscribe(){ Subscription subscription = RxBus.getInstance().doSubscribe(String.class, new Action1 () { @Override public void call(String s) { Toast.makeText(MainActivity.this,s, Toast.LENGTH_SHORT).show(); } }); RxBus.getInstance().addSubscription(this,subscription); } 或者
public void doSubscribe(){ Subscription subscribe = RxBus.getInstance().toObservable(String.class) .filter(new Func1 () { @Override public Boolean call(String s) { return true; } }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Action1 () { @Override public void call(String s) { Log.i(TAG, "call: " + s); } }); RxBus.getInstance().addSubscription(this,subscribe); } 取消订阅：
@Overrideprotected void onDestroy() { //取消订阅,释放内存 RxBus.getInstance().unSubscribe(this); super.onDestroy(); } 功能的完善
当我们使用PublishSubject时,可能有些功能还不是很完善,比如我们只能先订阅事件,然后发送事件,如果反过来,先发送了事件再进
行订阅操作，比如两个Activity之间传递消息，怎么保证发送的事件不丢失呢？也就是EventBus的StickEvent功能，这个时候
PublishSubject就没有办法实现，我们可以替换为BehaviorSubject
private RxBus() { mSubject = new SerializedSubject<>(BehaviorSubject.create()); } 但是BehaviorSubject只能缓存最近一个发送给它的事件，如果我们需要缓存多个事件可以改用ReplaySubject