手撸事件总线（初级版）

为什么抛弃传统的EventBus事件总线？

1，在Android开发中EventBus是非常流行的，能够解耦Handler、Thread、Broadcast以及AsyncTask等各种组件，并且可以跨Fragment通信。但这些前提都是需要合理使用，如果使用不合理，就会在代码中出现非常多的Event，导致代码混乱，维护起来让人没有头绪，这也是被人所诟病的地方。

2，RxJava响应式编程非常流行，如果项目中使用了RxJava就没必要使用EventBus，因为使用RxJava基本可以编写实现EventBus的全部功能，而且代码量极少。

初级版本的RxJava事件总线效果图？

代码实现

/**
 * author:Struggle
 * date:2019/6/12
 * email:zhuhuitao_struggle@163.com
 * description:
 */
public class RxBus {
    private final Subject<Object> mBus;
    private RxBus (){
        mBus = PublishSubject.create().toSerialized();
    }

    public static  getBus(){
        return Holder.bus;
    }

    public void post(Object obj){
        mBus.onNext(obj);
    }

    public <T>Observable<T> tObservable(Class<T> tClass){
        return mBus.ofType(tClass);
    }

    public Observable<Object> toObservable(){
        return mBus;
    }

    public boolean hasObservaable(){
        return  mBus.hasObservers();
    }

    private static class Holder{
        private static RxBus bus = new  RxBus();
    }
}

在这个基本的事件中，我们使用了Subject来发送事件，这哥们很厉害，充当双重角色，既是Observerable，又是Observer(Subscriber),它是整个事件中的重量级人物，关于Subject相关知识我们可以到https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
查阅。

在RxBus的私有构造函数中使用：

    private RxBus (){
        mBus = PublishSubject.create().toSerialized();
    }

这样的好处在于多线程环境下，可以保证线程安全，如果单独使用Subject是线程不安全的。

这个RxBus看上去是非常简单的，但是通过效果图确实可以看到它实现了EventBus基本的发布与订阅功能。

如何使用上面编写的RxBus？

1，首先对Activity做了一个基类封装：

public abstract class BaseActivity extends FragmentActivity {
    public RxBus rxBus;
    private ThreadPoolExecutor executors;
    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        rxBus = RxBus.getBus();
        setContentView(setLayoutId());
        createExecutors();
        findViews();
        initEvent();
        initDate();
    }

    /**
     * 创建线程池
     */
    protected  void createExecutors(){
        executors = new ThreadPoolExecutor(3, 30, 1, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
    }

    /**
     * 初始化事件
     */
    protected abstract void initEvent();

    /**
     * 查找控件
     */
    protected abstract void findViews();

    /**
     * 初始化数据
     */
    protected abstract void initDate();

    /**
     * 设置布局
     * @return
     */
    protected abstract int setLayoutId();

}

2,继承BaseActivity,使用RxBus发布与订阅事件：

public class MainActivity extends BaseActivity {
private CompositeDisposable mCompositeDisposable;
private final String TAG = "main";
private TextView mTvMessage;
private Button mBtSend;

    @Override
    protected void initEvent() {
        this.mBtSend.setOnClickListener(v->
            rxBus.post("Hello RxBus")//发布消息
        );
    }

    @Override
    protected void findViews() {
        this.mTvMessage = this.findViewById(R.id.tv_message);
        this.mBtSend = this.findViewById(R.id.bt_send);
    }

    @Override
    protected void initDate() {
        mCompositeDisposable = new CompositeDisposable();
        registerEvent();
    }

    @Override
    protected int setLayoutId() {
        return R.layout.activity_main;
    }

    //订阅消息
    private void registerEvent(){
        mCompositeDisposable
                .add(rxBus.tObservable(String.class)
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe((result ->
                            mTvMessage.setText("已接收到消息"+result)//更新UI
                        ))
        );
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        //取消所有的门阐，防止内存泄漏
        mCompositeDisposable.clear();
    }
}

首先需要注册Event，registerEvent()函数对事件进行了订阅，在注册事件中使用了CompositeDisposable,这个类在RxJava2.x中，具体可以到RxJava官方文档了解，在退出Activity时，调用了mCompositeDiaposable.clear()方法以便取消所有门阐，防止内存泄漏，另外我们将RxBus定义在BaseActivity中，最后使用rxBus.post(),发布消息。

至此一个初级版的事件总线已基本实现完成。