RxBus–打造極簡(jiǎn)
**原創(chuàng)CSDN **
Rxjava出來(lái)至今，在項(xiàng)目的使用中非常頻繁，被玩熟的功能也感覺越來(lái)越強(qiáng)大。如果你對(duì)Rxbus還有不熟的地方，這篇文章就帶你從實(shí)現(xiàn)的角度與使用的角度了解Rxbus,快速在項(xiàng)目中。
Rxbus到底是什么
一句話概括：用注冊(cè)訂閱者模式實(shí)現(xiàn)的通訊方式。
簡(jiǎn)簡(jiǎn)單單的一個(gè) 觀察者observable 與 Subscriber。
聽起來(lái)很簡(jiǎn)單，做起來(lái)更簡(jiǎn)單。
RxBus能在項(xiàng)目中做什么
它是以observable 與 Subscriber 實(shí)現(xiàn)的，正常注冊(cè)訂閱所傳遞的參數(shù)我們都是可以傳遞的。在項(xiàng)目中我們不例外的都是
Activity，F(xiàn)ragment 互相與各自間的傳遞 String Object！完全夠用是不是！傳遞對(duì)象的時(shí)候序列化對(duì)象也不是那么必要的。
是不是感覺很方便，一步post掉任何Object與String的參數(shù)。越厲害 越簡(jiǎn)單
我們?yōu)槭裁匆陧?xiàng)目中使用
1，當(dāng)我們項(xiàng)目已經(jīng)導(dǎo)入了Rxjava,RxAndroid 的時(shí)候，你自己創(chuàng)造的Rxbus能替換掉EventBus，瞬間發(fā)覺自己好牛是不是。而且項(xiàng)目中也少一個(gè)三方庫(kù)，Apk也瘦身咯。
2，方便易用，一個(gè)你自己完成的極簡(jiǎn)通訊方式，更不容易出現(xiàn)未知的bug。它的效率也是毋庸置疑的，很多大項(xiàng)目都是自己創(chuàng)建Rxbus取代EventBus的使用。
*RxBus的原理到底是什么
RxBus實(shí)現(xiàn)的方式有很多，整體上原理一直，細(xì)節(jié)略有不同，我重點(diǎn)說(shuō)下我的實(shí)現(xiàn)方式。
1，ConcurrentHashMap 一個(gè)與Hashmap,Hashtable 有點(diǎn)相似的集合。ConcurrentHashMap采用了分段鎖的設(shè)計(jì)，只有在同一個(gè)分段內(nèi)才存在競(jìng)態(tài)關(guān)系，不同的分段鎖之間沒有鎖競(jìng)爭(zhēng)。相比于對(duì)整個(gè)Map加鎖的設(shè)計(jì)，分段鎖大大的提高了高并發(fā)環(huán)境下的處理能力。但同時(shí)，由于不是對(duì)整個(gè)Map加鎖，導(dǎo)致一些需要掃描整個(gè)Map的方法（如size(),containsValue()）需要使用特殊的實(shí)現(xiàn)，另外一些方法（如clear()）甚至放棄了對(duì)一致性的要求。
它的效率科比Hashtable官方說(shuō)快了16倍！!
2，維護(hù)的這個(gè)集合可以看成我們observable 與subscriber。但我為了能能多個(gè)地方注冊(cè)，集合的第二參數(shù)是集合，里面存放的是subject.(subject也可以理解為observable,它的確是集成自被觀察者的)
3，集合有了，里面結(jié)構(gòu)已經(jīng)確定，剩下的步驟我們就是不停的往集合里面存數(shù)據(jù)與取數(shù)據(jù)了！
存 – 理解為訂閱 – 想集合中put一個(gè)subscriber
取 -- 理解創(chuàng)建observable -- 調(diào)用了集合中的subscriber的onnext的方法
是不是好簡(jiǎn)單。道理也就這個(gè)樣子了，具體結(jié)構(gòu)與邏輯的考慮我將在代碼中體現(xiàn)
完整的RxBus
結(jié)構(gòu)考慮得多了點(diǎn)，你可以根據(jù)自己的需要做一些變化。Rxbus本不難，有一個(gè)自己的rxbus想想多好！
/**
- 用RxJava實(shí)現(xiàn)的EventBus
- on 2017.03.23
- 先訂閱 再post 事件就能接受
*/
public class RxBus {
private static RxBus instance;
public static synchronized RxBus getInstance() {
if (null == instance) {
instance = new RxBus();
}
return instance;
}
private RxBus() {
}
@SuppressWarnings("rawtypes")
private ConcurrentHashMap<Object, List<Subject>> subjectMapper = new ConcurrentHashMap<Object, List<Subject>>();
/**
* 注冊(cè)事件源
*
* @param tag
* @return
*/
@SuppressWarnings({"rawtypes"})
public <T> Observable<T> register(@NonNull Object tag) {
List<Subject> subjectList = subjectMapper.get(tag);
if (null == subjectList) {
subjectList = new ArrayList<Subject>();
subjectMapper.put(tag, subjectList);
}
Subject<T, T> subject;
subjectList.add(subject = PublishSubject.create());
LogUtils.logd("register" + tag + " size:" + subjectList.size());
return subject;
}
@SuppressWarnings("rawtypes")
public void unregister(@NonNull Object tag) {
List<Subject> subjects = subjectMapper.get(tag);
if (null != subjects) {
subjectMapper.remove(tag);
}
}
/**
* 取消監(jiān)聽
*
* @param tag
* @param observable
* @return
*/
@SuppressWarnings("rawtypes")
public RxBus unregister(@NonNull Object tag,
@NonNull Observable<?> observable) {
if (null == observable)
return getInstance();
List<Subject> subjects = subjectMapper.get(tag);
if (null != subjects) {
subjects.remove((Subject<?, ?>) observable);
if (isEmpty(subjects)) {
subjectMapper.remove(tag);
LogUtils.logd("unregister" + tag + " size:" + subjects.size());
}
}
return getInstance();
}
public void post(@NonNull Object content) {
post(content.getClass().getName(), content);
}
/**
* 觸發(fā)事件
*
* @param content
*/
@SuppressWarnings({"unchecked", "rawtypes"})
public void post(@NonNull Object tag, @NonNull Object content) {
LogUtils.logd("post" + "eventName: " + tag);
List<Subject> subjectList = subjectMapper.get(tag);
if (!isEmpty(subjectList)) {
for (Subject subject : subjectList) {
subject.onNext(content);
LogUtils.logd("onEvent" + "eventName: " + tag);
}
}
}
/*
* Subject 是Observable子類
* */
@SuppressWarnings("rawtypes")
public static boolean isEmpty(Collection<Subject> collection) {
return null == collection || collection.isEmpty();
}
}
RxBus也就這么多的點(diǎn)了
