spring中配置EventBus

1. 概述

       guava中EventBus是一个消息处理总线，基于观察者模式设计和实现。网上有很多关于其原理介绍和用法的文字，但很多用法都是给了已经简单的示例或测试demo，并没有和工程框架相结合。

      本文主要介绍下在spring框架中如何使用和配置EventBus。

2. EventBus在spring中的使用

      EventBus主要分为两种，一种是同步消息总线（EventBus）；另一种是异步消息总线（AsyncEventBus）。

     下面主要以同步消息总线为例来介绍下EventBus和Spring的结合使用。

2.1 定义EventBus工具类

      guava中的com.google.common.eventbus.EventBus提供了很多API，在spring中使用时最好能够定义了bean，方便同一个消息总线对象被多处功能公用，同时也便于代码维护。

public class TimeEventBus {

/** 时间任务总线 */
private final static EventBus tiemEventBus = new EventBus();

/**
* 触发同步事件
*
* @param event
*/
public static void post(Object event) {
tiemEventBus.post(event);
}

/**
* 注册事件处理器
*
* @param handler
*/
public static void register(Object handler) {
tiemEventBus.register(handler);
}

/**
* 注销事件处理器
*
* @param handler
*/
public static void unregister(Object handler) {
tiemEventBus.unregister(handler);
}
}

      TimeEventBus类封装了guava原始EventBus的api，并通过static方法对外提供常用的消费发送、消息订阅、取消订阅等方法。

2.2 定义消息体

        一个消息总线可以支持多种消息的订阅和发送，不同的消息通过类名来区分。

消息体使用一个简单的Java DO，本身不需要继承或实现EventBus的接口。例如定义一个时间过期的消息实例如下：

public class TimeExpiringEvent {

private TimeTask timeTask;

public TimeExpiringEvent(TimeTask lifeTimeTask){
this.timeTask = lifeTimeTask;
}

public TimeTask getLifeTimeTask() {
return timeTask;
}

public void setLifeTimeTask(TimeTask lifeTimeTask) {
this.timeTask = lifeTimeTask;
}
}

2.3 定义消息监听器类

    消息监听器用于注册、监听和处理从消息总线接收到订阅消息，代码如下：

public class TimeTaskEventListener {

@PostConstruct
public void init() {
TimeEventBus.register(this);
}

@Subscribe
public boolean processExpiringEvent(TimeExpiringEvent expiringEvent) {

//TODO process code

return true;
}
}

    消息总线的注册在init()方法中实现，消息的订阅则通过@Subscribe注解方式实现。

2.4 spring中配置消息监听器

    为了在spring容器启动后完成监听器在EventBus中的注册，需要在spring中配置监听器的bean实体。

<bean id="timeTaskEventListener" class="com.cc.guava.event.listener.TimeTaskEventListener" />

2.5 消息发送

    最后就差一步消息发送方的代码实现了，消息发送通过post方法实现。

TimeEventBus.post(new TimeExpiringEvent(timeTask));

3. 异步消息总线 AsyncEventBus 使用

       AsyncEventBus的大部分使用方式和同步消息总线类型，唯一不同的是，需要对异步消息的线程池实现进行定义，下面给出了默认缓存池配置的代码。

private final static AsyncEventBus timeAsyncEventBus = new AsyncEventBus(Executors.newCachedThreadPool());

    在使用时需要根据自己的业务特点，选择相应的多线程缓存池配置。