Guava: 事件总线EventBus

时间:2021-08-21 22:48:07
Guava: 事件总线EventBus

EventBus 直译过来就是事件总线,它使用发布订阅模式支持组件之间的通信,不需要显式地注册回调,比观察者模式更灵活,可用于替换Java中传统的事件监听模式,EventBus的作用就是解耦,它不是通用的发布订阅系统,也不能用于进程间通信。可用于Android的EventBus库主要有这几个:Google出品的Guava,Guava是一个庞大的库,EventBus 只是它附带的一个小功能,因此实际项目中使用并不多。用的最多的是greenrobot/EventBus,这个库的优点是接口简洁,集成方便,但是限定了方法名,不支持注解。另一个库square/otto修改自 Guava ,用的人也不少。

以greenrobot/EventBus 为例,我们看一下 EventBus 模式的典型用法:

// 注册EventBus,接受事件
class Fragment {
public void onCreate(){
EventBus.getDefault().register(this);
}
public void onDestroy(){
EventBus.getDefault().unregister(this);
}
public void onEvent(SomeEvent1 event){
// handle event
}
} // 处理任务,发送事件
public class Service {
public void doSomeThing(){
// do your work
// send event
EventBus.getDefault().post(new SomeEvent1());
}

关于EventBus中的几个问题?

  1. 事件定义:任意的对象即可;
  2. 事件处理器的注册:事件处理的方法,添加注解即可,然后事件处理器的对象注册到总线中,总线维护一个事件和事件处理器的关联关系,在内存中;
  3. 事件的处理过程:同步处理和异步处理,事件提交之后,事件队列维护在本地缓存,同步的方式直接当前线程去执行,异步的处理策略是在初始化事件总线的时候就搞了一个线程池出来,由线程池去异步执行;
  4. EventBus就开放了三个方法,register/post/unregister
  5. 为什么会有unregister?在99.99%的使用场景中,是不会在runtime的时候去register/unregister某个observer的,在spring的环境,也是在init的时候做register/unregister。不过做framework就必须要考虑这0.01%的使用场景。

一、Guava EventBus 观察者模式

首先,我们声明一个Observer:

public class EventObserver {
@Subscribe public void onMessage(Message message) {
...
}
}

这个类并没有继承任何接口,只是在用来响应通知的方法上声明了一个@Subscribe。

使用EventBus很简单,先声明一个:

EventBus eventBus = new EventBus();

然后,把我们写好的Observer注册进去:

eventBus.register(new EventObserver());

当要通知Observer时,我们只要这样即可:

eventBus.post(message);

这里,我们并没有告诉EventBus,我们要处理的是一个Message类型,只是在EventObserver的onMessage方法的接口声明上使用了这个类型而已。但是,当我们把消息发送出去的时候,它会根据类型进行匹配,保证我们的消息正确地发送到对应的地方。

相比于JDK原有的实现,这个实现会更简单。EventObserver不再需要存在一个继承体系中,而继承总是一种枷锁,把我们套牢在一个体系之中:

  • 我们不必遵循一个特定的名字,比如Observer的update,而这里的名字onMessage是我们自己起的。
  • 我们不必遵循特定的类型,比如update方法中作为被观察对象Observable和作为参数的Object,而是根据我们自己的需求选择的类型。

这种变换让静态类型的Java语言,有了一些动态类型的特质,也让程序更加灵活。这种灵活性多半要归功于Annotation,它在很大程度上影响了Java的程序设计风格。

除了标准的EventBus,Guava还提供了另外一个AsyncEventBus,从名字就可以看出,这是一个异步的EventBus,也就是说,消息扔给它之后,会立即返回,至于Observer什么时候处理,那就是它的事情了。当处理耗时的处理时很有用,我们要依赖Executors来实现异步事件总线。

AsyncEventBus eventBus = new AsyncEventBus("test", Executors.newCachedThreadPool());

另外:关于EventBus的使用请参见:http://blog.mcxiaoke.com/2015/08/03/how-to-write-an-eventbus-part1/

二、示例

1、一个事件的定义(任何对象都可以是事件)

public class SignEvent {    

    private String companyName; 

    private String signName;    

    private Date signDate; 

    public SignEvent(String name,String signName, Date signDate) { 

        super(); 

        this.companyName = name; 

        this.signName = signName; 

        this.signDate = signDate; 

    }   

    public String getMessage(){ 

        StringBuilder sb = new StringBuilder(); 

        sb.append("物流公司:").append(this.companyName); 

        sb.append("签收人:").append(signName).append(",签收日期:").append(signDate); 

        return sb.toString(); 

    } 

}

2、定义两个事件监听器,添加注解做事件的订阅

public class YTOEventListener { 

    @Subscribe 

    public void consign(SignEvent signEvent){ 

        if(signEvent.getCompanyName().equalsIgnoreCase("YTO")){ 

            System.out.println("YTO。。。开始发货"); 

            System.out.println(signEvent.getMessage()); 

        } 

    } 

    @Subscribe 

    public void delivery(SignEvent signEvent){ 

        if(signEvent.getCompanyName().equalsIgnoreCase("YTO")){ 

            System.out.println("YTO。。。开始投递"); 

        } 

    } 

}

public class SFEventListener {  

    @Subscribe 

    public void consign(SignEvent signEvent){ 

        if(signEvent.getCompanyName().equalsIgnoreCase("SF")){ 

            System.out.println("SF。。。开始发货"); 

            System.out.println(signEvent.getMessage()); 

        } 

    }   

    @Subscribe 

    public void delivery(SignEvent signEvent){ 

        if(signEvent.getCompanyName().equalsIgnoreCase("SF")){ 

            System.out.println("SF。。。开始投递"); 

        } 

    } 

}

3、EventBus的例子,包含时间的注册以及事件的提交

public class EventBusTest { 

    public static void siginalThreadConsumer(){ 

        EventBus bus = new EventBus("iamzhongyong");        

        SFEventListener sf = new SFEventListener(); 

        YTOEventListener yto = new YTOEventListener(); 

        bus.register(sf); 

        bus.register(yto);      

        SignEvent sign1 = new SignEvent("SF","比熊啊",new Date()); 

        bus.post(sign1);        

        SignEvent sign2 = new SignEvent("YTO","你妹的",new Date()); 

        bus.post(sign2);    

    } 

    public static void multiThread(){ 

        EventBus bus = new AsyncEventBus(Executors.newFixedThreadPool(3));      

        SFEventListener sf = new SFEventListener(); 

        YTOEventListener yto = new YTOEventListener(); 

        bus.register(sf); 

        bus.register(yto); 

        SignEvent sign1 = new SignEvent("SF","比熊啊",new Date()); 

        bus.post(sign1);        

        SignEvent sign2 = new SignEvent("YTO","你妹的",new Date()); 

        bus.post(sign2);    

    } 

    public static void main(String[] args) {        

        EventBusTest.siginalThreadConsumer(); 

        EventBusTest.multiThread(); 

    } 

}