RxBus–打造极简
**原创CSDN **
Rxjava出来至今，在项目的使用中非常频繁，被玩熟的功能也感觉越来越强大。如果你对Rxbus还有不熟的地方，这篇文章就带你从实现的角度与使用的角度了解Rxbus,快速在项目中。
Rxbus到底是什么
一句话概括：用注册订阅者模式实现的通讯方式。
简简单单的一个 观察者observable 与 Subscriber。
听起来很简单，做起来更简单。
RxBus能在项目中做什么
它是以observable 与 Subscriber 实现的，正常注册订阅所传递的参数我们都是可以传递的。在项目中我们不例外的都是
Activity，Fragment 互相与各自间的传递 String Object！完全够用是不是！传递对象的时候序列化对象也不是那么必要的。
是不是感觉很方便，一步post掉任何Object与String的参数。越厉害 越简单
我们为什么要在项目中使用
1，当我们项目已经导入了Rxjava,RxAndroid 的时候，你自己创造的Rxbus能替换掉EventBus，瞬间发觉自己好牛是不是。而且项目中也少一个三方库，Apk也瘦身咯。
2，方便易用，一个你自己完成的极简通讯方式，更不容易出现未知的bug。它的效率也是毋庸置疑的，很多大项目都是自己创建Rxbus取代EventBus的使用。
*RxBus的原理到底是什么
RxBus实现的方式有很多，整体上原理一直，细节略有不同，我重点说下我的实现方式。
1，ConcurrentHashMap 一个与Hashmap,Hashtable 有点相似的集合。ConcurrentHashMap采用了分段锁的设计，只有在同一个分段内才存在竞态关系，不同的分段锁之间没有锁竞争。相比于对整个Map加锁的设计，分段锁大大的提高了高并发环境下的处理能力。但同时，由于不是对整个Map加锁，导致一些需要扫描整个Map的方法（如size(),containsValue()）需要使用特殊的实现，另外一些方法（如clear()）甚至放弃了对一致性的要求。
它的效率科比Hashtable官方说快了16倍！!
2，维护的这个集合可以看成我们observable 与subscriber。但我为了能能多个地方注册，集合的第二参数是集合，里面存放的是subject.(subject也可以理解为observable,它的确是集成自被观察者的)
3，集合有了，里面结构已经确定，剩下的步骤我们就是不停的往集合里面存数据与取数据了！
存 – 理解为订阅 – 想集合中put一个subscriber
取 -- 理解创建observable -- 调用了集合中的subscriber的onnext的方法
是不是好简单。道理也就这个样子了，具体结构与逻辑的考虑我将在代码中体现
完整的RxBus
结构考虑得多了点，你可以根据自己的需要做一些变化。Rxbus本不难，有一个自己的rxbus想想多好！
/**
- 用RxJava实现的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>>();
/**
* 注册事件源
*
* @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);
}
}
/**
* 取消监听
*
* @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);
}
/**
* 触发事件
*
* @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也就这么多的点了
