大家好涕刚，我叫石頭
前言
事件總線出現(xiàn)的原因：為了使組件之間的通信變得簡單嗡综，深度解耦！
說白了就是切斷組件之間的直接聯(lián)系副女，采用 發(fā)布/訂閱 的模式（觀察者模式）
相信我們很多人都用過EventBus或者Otto來作為我們APP中的事件總線蛤高，所以我們會有這樣的困惑蚣旱，RxBus真的能替代EventBus嗎？
那接下來我們就先來分析分析下:
This project is deprecated in favor of RxJava and RxAndroid. These projects permit the same event-driven programming model as Otto, but they’re more capable and offer better control of threading.
該項目已被RxJava和RxAndroid取代戴陡。Rx類項目允許與Otto類似的事件驅(qū)動編程模型塞绿，而且能力更強，操作線程更方便恤批。
Otto已經(jīng)停止開發(fā)了异吻，所以我們只需對比EventBus和RxBus了。
對于EventBus和RxBus的比較我們要先明白 一個完美的事件總線應該具備哪些功能喜庞？
容易訂閱事件：事件訂閱者只要聲明自己就好了诀浪，當事件發(fā)生時自然會被調(diào)到。訂閱和取消可以方便綁定到Activity和Fragment的生命周期上延都。
容易發(fā)送事件：事件發(fā)送者直接發(fā)送就好了雷猪，其他的事都不管。
方便的切換線程：有些事必須主線程干晰房，有些事必須非主線程干求摇，所以這個還是要說清楚。
性能：隨著應用的成長殊者，總線可能會被重度使用与境，性能一定要好。
糾結(jié)到底是用EventBus還是RxBus的朋友可以參考這篇文章--RxBus真的能替代EventBus嗎猖吴？
接下來我們就開始RxBus之旅了---------------
一摔刁、添加RxJava和RxAndroid依賴
//RxJava and RxAndroid
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
順便說下，我們是用的rxjava1.X的版本海蔽，現(xiàn)在也有了rxjava2.x的版本共屈，他們之間有些區(qū)別，感興趣的朋友可以去看看准潭。
二趁俊、建立RxBus類
import java.util.HashMap;
import rx.Observable;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subscriptions.CompositeSubscription;
/**
* Created by shitou on 2017/4/26.
*/
public class RxBus {
private static volatile RxBus mInstance;
/**
* PublishSubject只會把在訂閱發(fā)生的時間點之后來自原始Observable的數(shù)據(jù)發(fā)射給觀察者
*/
private SerializedSubject<Object, Object> mSubject;
private HashMap<String, CompositeSubscription> mSubscriptionMap;
private RxBus() {
mSubject = new SerializedSubject<>(PublishSubject.create());
}
public static RxBus getInstance() {
if (mInstance == null) {
synchronized (RxBus.class) {
if (mInstance == null) {
mInstance = new RxBus();
}
}
}
return mInstance;
}
/**
* 發(fā)送事件
*/
public void post(Object o) {
mSubject.onNext(o);
}
/**
* 是否已有觀察者訂閱
*/
public boolean hasObservers() {
return mSubject.hasObservers();
}
/**
* 一個默認的訂閱方法
*/
public <T> Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> error) {
return toObservable(type)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(next, error);
}
/**
* 返回指定類型的Observable實例
*/
public <T> Observable<T> toObservable(final Class<T> type) {
return mSubject.ofType(type);
}
/**
* 保存訂閱后的subscription
*/
public void addSubscription(Object o, Subscription subscription) {
if (mSubscriptionMap == null) {
mSubscriptionMap = new HashMap<>();
}
String key = o.getClass().getName();
if (mSubscriptionMap.get(key) != null) {
mSubscriptionMap.get(key).add(subscription);
} else {
CompositeSubscription compositeSubscription = new CompositeSubscription();
compositeSubscription.add(subscription);
mSubscriptionMap.put(key, compositeSubscription);
}
}
/**
* 取消訂閱
*/
public void unSubscribe(Object o) {
if (mSubscriptionMap == null) {
return;
}
String key = o.getClass().getName();
if (!mSubscriptionMap.containsKey(key)){
return;
}
if (mSubscriptionMap.get(key) != null) {
mSubscriptionMap.get(key).unsubscribe();
}
mSubscriptionMap.remove(key);
}
}
在RxJava中有個Subject類，它繼承Observable類刑然，同時實現(xiàn)了Observer接口，因此Subject可以同時擔當訂閱者和被訂閱者的角色暇务，這里我們使用Subject的子類PublishSubject來創(chuàng)建一個Subject對象（PublishSubject只有被訂閱后才會把接收到的事件立刻發(fā)送給訂閱者）
Rxjava中泼掠，訂閱操作會返回一個Subscription對象，以便在合適的時機取消訂閱垦细，防止內(nèi)存泄漏择镇，如果一個類產(chǎn)生多個Subscription對象，我們可以用一個CompositeSubscription存儲起來括改，以進行批量的取消訂閱腻豌。
由于Subject類是非線程安全的，所以我們通過它的子類SerializedSubject將PublishSubject轉(zhuǎn)換成一個線程安全的Subject對象。
public <T> Observable<T> toObservable(final Class<T> type) {
return mSubject.ofType(type);
}
ofType()方法能過濾掉不符合條件的事件類型（比如你的type是EventType1.class,那么就只能輸出EventType1.class的類型）吝梅，然后將滿足條件的事件類型通過cast()方法虱疏，轉(zhuǎn)換成對應類型的Observable對象，這是在源碼中轉(zhuǎn)換的苏携。
/**
* 一個默認的訂閱方法
*/
public <T> Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> error) {
return toObservable(type)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(next, error);
}
上面的方法封裝了訂閱方法做瞪，并且指定了執(zhí)行線程，我們只需要傳入type（事件類型）右冻，next（成功的Action1）装蓬，error(錯誤的Action1)，其實你也可以根據(jù)你自己的需要封裝自己的doSubscribe方法纱扭，來簡化代碼牍帚。
在需要發(fā)送事件的地方調(diào)用post()方法，它間接的通過mSubject.onNext(o);將事件發(fā)送給訂閱者乳蛾。
同時RxBus提供了addSubscription()暗赶、unSubscribe()方法，分別來保存訂閱時返回的`Subscription對象屡久，以及取消訂閱忆首。
實戰(zhàn)一
主線程(UI線程)發(fā)送String類型的事件
button點擊事件代碼
mButton1 = (Button) findViewById(R.id.button);
mButton1.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
//在主線程中發(fā)送String類型的事件
RxBus.getInstance().post("hello RxBus!");
}
});
在onCreate中實現(xiàn)下面代碼
Subscription subscription = RxBus.getInstance()
.toObservable(String.class)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
mTextView.setText("接收的事件內(nèi)容"+s);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
Log.e(TAG, "error");
}
});
之后我們可以把subscription對象保存到HashMap<String, CompositeSubscription>集合中去。
RxBus.getInstance().addSubscription(this,subscription);
這樣當我們點擊button時被环，textview就收到了消息糙及。
最后，一定要記得在生命周期結(jié)束的地方取消訂閱事件筛欢，防止RxJava可能會引起的內(nèi)存泄漏問題浸锨。
protected void onDestroy() {
super.onDestroy();
RxBus.getInstance().unSubscribe(this);
}
實戰(zhàn)二
在子線程中發(fā)送Integer類型的事件
button點擊事件代碼
mButton2 = (Button) findViewById(R.id.button2);
mButton2.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
new Thread(new Runnable() {
@Override
public void run() {
RxBus.getInstance().post(1234);
}
}).start();
}
});
在onCreate中實現(xiàn)下面代碼
Subscription subscription1 = RxBus.getInstance()
.doSubscribe(Integer.class, new Action1<Integer>() {
@Override
public void call(Integer integer) {
mTextView.setText("接收的事件內(nèi)容"+integer);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
Log.e(TAG, "error");
}
});
之后我們可以把subscription對象保存到HashMap<String, CompositeSubscription>集合中去。
RxBus.getInstance().addSubscription(this,subscription1);
最后版姑，一定要記得在生命周期結(jié)束的地方取消訂閱事件柱搜，防止RxJava可能會引起的內(nèi)存泄漏問題。
protected void onDestroy() {
super.onDestroy();
RxBus.getInstance().unSubscribe(this);
}
上面的都是發(fā)送的基本數(shù)據(jù)類型剥险，那么我們能不能發(fā)送自己封裝的類型呢聪蘸？答案是：肯定行的拗胜！
實戰(zhàn)三
創(chuàng)建你要發(fā)送的事件類
下面我們來創(chuàng)建一個學生類:StudentEvent
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;
}
}
發(fā)送事件
RxBus.getInstance().post(new StudentEvent("110","小明"));
注冊和接收事件
Subscription subscription2 = RxBus.getInstance()
.toObservable(StudentEvent.class)
.observeOn(Schedulers.io())
.subscribeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<StudentEvent>() {
@Override
public void call(StudentEvent studentEvent) {
String id = studentEvent.getId();
String name = studentEvent.getName();
mTextView.setText("學生的id:"+id+" 名字:"+name);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
}
});
最后第煮，一定要記得在生命周期結(jié)束的地方取消訂閱事件，防止RxJava可能會引起的內(nèi)存泄漏問題幔欧。
protected void onDestroy() {
super.onDestroy();
RxBus.getInstance().unSubscribe(this);
}
實戰(zhàn)四
廣播中發(fā)送事件么介，訂閱方式按照實戰(zhàn)一的方式娜遵。
定義一個檢測網(wǎng)絡狀態(tài)的廣播：
public class NetworkChangeReceiver extends BroadcastReceiver {
@Override
public void onReceive(Context context, Intent intent) {
ConnectivityManager manager = (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);
NetworkInfo networkInfo = manager.getActiveNetworkInfo();
if (networkInfo != null && networkInfo.isAvailable()) {
RxBus.getInstance().post("網(wǎng)絡連接成功");
} else {
RxBus.getInstance().post("網(wǎng)絡不可用");
}
}
}
在網(wǎng)絡可用與不可用時發(fā)送提示事件，然后在onCreate()方法中注冊廣播：
private void registerReceiver() {
IntentFilter intentFilter = new IntentFilter();
intentFilter.addAction("android.net.conn.CONNECTIVITY_CHANGE");
mReceiver = new NetworkChangeReceiver();
registerReceiver(mReceiver, intentFilter);
}
最后不要忘了在onDestory()中對廣播進行取消注冊壤短，以及取消訂閱设拟。
protected void onDestroy() {
super.onDestroy();
unregisterReceiver(mReceiver);
RxBus.getInstance().unSubscribe(this);
}
到這里我們實現(xiàn)了幾種事件傳送慨仿，但是細心的童鞋可能發(fā)現(xiàn)我們在上面的例子中都是先訂閱 事件,然后發(fā)送 事件(因為我們是用的PublishSubject,PublishSubject只會把在訂閱發(fā)生的時間點之后來自原始Observable的數(shù)據(jù)發(fā)射給觀察者，這在前面我們提到過),如果我們反過來纳胧，先發(fā)送了事件镰吆，再進行訂閱操作，怎么保證發(fā)送的事件不丟失呢躲雅？也就是EventBus中的StickyEven功能鼎姊。RxBus--支持Sticky事件里面講解了Subject的4種實現(xiàn)，有興趣的朋友可以去看看相赁。
最后推薦一些RxJava的學習資源：RxJava入門相寇、給 Android 開發(fā)者的 RxJava 詳解
