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<Object,Object> mSubject;
 28     //一个类产生多个Subscription对象，用一CompositeSubscription 存储起来，以进行批量的取消订阅。避免内存泄漏
 29     private HashMap<String,CompositeSubscription> 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 <T>
 55      * @return
 56      */
 57     public <T>Observable<T> toObservable(final Class<T> 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 <T>
 72      * @return
 73      */
 74     public <T>Subscription doSubscribe(Class<T> type, Action1<T> next){
 75         return toObservable(type)
 76                 .subscribeOn(Schedulers.io())
 77                 .observeOn(AndroidSchedulers.mainThread())
 78                 .subscribe(next);
 79     }
 80     /**
 81      * 默认的订阅方法
 82      * @param <T>
 83      * @return
 84      */
 85     public <T>Subscription doSubscribe(Class<T> type, Action1<T> next,Action1<Throwable> error){
 86         return toObservable(type)
 87                 .subscribeOn(Schedulers.io())
 88                 .observeOn(AndroidSchedulers.mainThread())
 89                 .subscribe(next,error);
 90     }
 91     /**
 92      * 默认的订阅方法
 93      * @param <T>
 94      * @return
 95      */
 96     public <T>Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> 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 o
106      * @param subscription
107      */
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 o
125      */
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 }