Android 事件总线

应用内消息传递

对于Android系统来说，消息传递是最基本的组件，每一个App内的不同页面，不同组件都在进行消息传递。消息传递既可以用于Android四大组件之间的通信，也可用于异步线程和主线程之间的通信。对于Android开发者来说，经常使用的消息传递方式有很多种，从最早使用的Handler、BroadcastReceiver、接口回调，到近几年流行的通信总线类框架EventBus、RxBus。Android消息传递框架，总在不断的演进之中。

什么是事件总线呢？

在简单观察模式中，观察者订阅被观察者，单被观察者状态或者数据发生变化时通知观察者，这是一对一的关系。
但当观察者和被观察者是多个或者不确定数量的时候，这就需要一个总线来存储这些观察者和被观察者，方便在发送通知的时候找到对应的观察者。

常用的事件总线方案

EventBus

详情见另一篇文章Eventbus源码分析

Rxbus

RxBus不是一个库，而是一个文件，实现只有短短30行代码。RxBus本身不需要过多分析，它的强大完全来自于它基于的RxJava技术。
Rxbus属于Rxjava库下一小部分功能，如果项目已经使用Rxjava，就不需要再额外引入EventBus库来使用了，直接使用Rxbus就可以了撒。需要从Eventbus转到Rxbus使用的童鞋可以放心转，使用方式大同小异。

引入库：

    implementation 'io.reactivex:rxjava:1.1.0'
    implementation 'io.reactivex:rxandroid:1.1.0'

创建Rxbus操作类：

Subject是非线程安全的，在并发情况下，不推荐使用通常的Subject对象，而是推荐使用SerializedSubject。

public class RxBus {
    private static volatile RxBus instance;
    private Subject<Object, Object> bus;

    private RxBus() {
        bus = new SerializedSubject<>(PublishSubject.create());
    }

    public static RxBus getDefault() {
        if (instance == null) {
            synchronized (RxBus.class) {
                instance = new RxBus();
            }
        }
        return instance;
    }

    /**
     * 发送事件
     * @param object
     */
    public void post(Object object) {
        bus.onNext(object);
    }

    /**
     * 根据类型接收相应类型事件
     * @param eventType
     * @param <T>
     * @return
     */
    public <T> Observable toObservable(Class<T> eventType) {
        return bus.ofType(eventType);
    }
}

在BaseActivity中保存和取消订阅事件。

public class BaseActivity extends AppCompatActivity {
    protected ArrayList<Subscription> rxBusList = new ArrayList<>();

    @Override
    protected void onDestroy() {
        super.onDestroy();

        clearSubscription();
    }

    /**
     * 取消该页面所有订阅
     */
    private void clearSubscription() {
        for (Subscription subscription : rxBusList) {
            if (subscription != null && subscription.isUnsubscribed()) {
                subscription.unsubscribe();
            }
        }
    }
}

发送事件方式：

RxBus.getDefault().post(new EventBean(1, "听说名字长回头率很高"));

接收事件方式：

        Subscription subscription = RxBus.getDefault().toObservable(EventBean.class)
            .subscribe(new Action1<EventBean>() {
            @Override
            public void call(EventBean eventBean) {
                tvContent.setText(eventBean.getUserId() + "------" + eventBean.getNickName());
            }
        });
        rxBusList.add(subscription);

Rxbus原理

这里是通过Rxjava中的PublishSubject.create().toSerialized() 来创建总线用来存储观察者。简单的就把它当做集合吧。

RxBus工作流程图

1、首先创建一个可同时充当Observer和Observable的Subject；
2、在需要接收事件的地方，订阅该Subject（此时Subject是作为Observable），在这之后，一旦Subject接收到事件，立即发射给该订阅者；
3、在我们需要发送事件的地方，将事件post至Subject，此时Subject作为Observer接收到事件（onNext），然后会发射给所有订阅该Subject的订阅者。

LiveDataBus

为何使用liveData做事件总线？

LiveData具有的这种可观察性和生命周期感知的能力，使其非常适合作为Android通信总线的基础构件。在一对多的场景中，发布消息事件后，订阅事件的页面只有在可见的时候才会处理事件逻辑。
使用者不用显示调用反注册方法。LiveData具有生命周期感知能力，所以LiveDataBus只需要调用注册回调方法，而不需要显示的调用反注册方法。这样带来的好处不仅可以编写更少的代码，而且可以完全杜绝其他通信总线类框架（如EventBus、RxBus）忘记调用反注册所带来的内存泄漏的风险。

LiveDataBus的组成

消息：消息可以是任何的 Object，可以定义不同类型的消息，如 Boolean、String。也可以定义自定义类型的消息。
消息通道： LiveData 扮演了消息通道的角色，不同的消息通道用不同的名字区分，名字是 String 类型的，可以通过名字获取到一个 LiveData 消息通道。
消息总线：消息总线通过单例实现，不同的消息通道存放在一个 HashMap 中。
订阅：订阅者通过 getChannel() 获取消息通道，然后调用 observe() 订阅这个通道的消息。
发布：发布者通过 getChannel() 获取消息通道，然后调用 setValue() 或者 postValue() 发布消息。

代码实现：

public final class LiveDataBus {

    private final Map<String, MutableLiveData<Object>> bus;

    private LiveDataBus() {
        bus = new HashMap<>();
    }

    private static class SingletonHolder {
        private static final LiveDataBus DATA_BUS = new LiveDataBus();
    }

    public static LiveDataBus get() {
        return SingletonHolder.DATA_BUS;
    }

    public <T> MutableLiveData<T> getChannel(String target, Class<T> type) {
        if (!bus.containsKey(target)) {
            bus.put(target, new MutableLiveData<T>());
        }
        return (MutableLiveData<T>) bus.get(target);
    }

    public MutableLiveData<Object> getChannel(String target) {
        return getChannel(target, Object.class);
    }
}

发送/接收消息：

//发送消息
LiveDataBus.get().getChannel("mykey").setValue(text);

//接收消息
LiveDataBus.get().getChannel("mykey", String.class)
        .observe(this, new Observer<String>() {
            @Override
            public void onChanged(@Nullable String newText) {
                // 更新数据
                tvText.setText(newText);
            }
        });

做法就是在单例的LiveDataBus中，使用Hashmap存储，key为自定义的String，value为MutablelivaData，通过发送和接收时候将key-value进行存取。注册利用了MutablelivaData的observer注册观察者。传递的消息对象类作为泛型，与MutablelivaData的泛型一致。

LiveDataBus遇到的问题和分析思路

1.订阅者会收到订阅之前发布的消息，类似于粘性消息。对于一个消息总线来说，这是不可接受的。
2.多次调用了 postValue() 方法，只有最后次调用的值会得到更新。也就是此方法是有可能会丢失事件！

FlowEventBus

MutableSharedFlow作为事件载体：
优点：
依托协程轻松切换线程
可以通过replay实现粘性效果
可以被多个观察者订阅
无观察者自动清除事件不会造成积压