Java—事件驱动进行代码解耦（EventBus）

阅读完需：约 3 分钟

虽然现在的各种应用都是集群部署，单机部署的应用越来越少了，但是不可否认的是，市场上还是存在许多单机应用的。本文要介绍的是 Guava 中的 EventBus 的使用。

EventBus 处理的事情类似观察者模式，基于事件驱动，观察者们监听自己感兴趣的特定事件，进行相应的处理。

添加 Guava 依赖

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>29.0-jre</version>
        </dependency>

定义一个注解用于标记 listener

/**
 * 用于标记 listener
 */
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface EventBusListener {
}

定义注册中心

/**
 * 事件注册中心
 * @author xujiahui
 */
@Component
public class EventBusCenter {

    final Integer AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();

    ThreadLocal<Exception> threadLocalSync = new ThreadLocal<>();

    /**
     *  管理同步事件
     */
    private final EventBus syncEventBus = new EventBus(
            (exception, context) -> threadLocalSync.set((Exception) exception)
    );

    /**
     * 管理异步事件
     */
    private final AsyncEventBus asyncEventBus = new AsyncEventBus(
            new ThreadPoolExecutor(AVAILABLE_PROCESSORS, 3 * AVAILABLE_PROCESSORS,
            10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(20),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy())
    );

    public void postSync(Object event) throws Exception {
        syncEventBus.post(event);
        Exception ex=threadLocalSync.get();
        if (ex != null) {
            // 记得 remove
            threadLocalSync.remove();
            throw ex;
        }
    }

    public void postAsync(Object event) throws Exception {
        asyncEventBus.post(event);
    }

    @PostConstruct
    public void init() {
        // 获取所有带有 @EventBusListener 的 bean，将他们注册为监听者
        List<Object> listeners = SpringContextUtils.getBeansWithAnnotation(EventBusListener.class);
        for (Object listener : listeners) {
            asyncEventBus.register(listener);
            syncEventBus.register(listener);
        }
    }
}

定义各种事件

举个例子，我们定义一个订单创建事件：

@Data
public class OrderCreatedEvent {
    private long orderId;
    private long userId;
    public OrderCreatedEvent(long orderId, long userId) {
        this.setOrderId(orderId);
        this.setUserId(userId);
    }
}

定义事件监听器

首先，类上面需要加我们之前定义的注解：@EventBusListener，然后监听方法需要加注解 @Subscribe，方法参数为具体事件。

监听器会根据参数类型来执行方式

@Component
@EventBusListener
public class OrderChangeListener {

    @Subscribe
    public void created(OrderCreatedEvent event) {
        long orderId = event.getOrderId();
        long userId = event.getUserId();
        // 订单创建成功后的各种操作，如发短信、发邮件等等。
        // 注意，事件可以被订阅多次，也就是说可以有很多方法监听 OrderCreatedEvent 事件，
        // 所以没必要在一个方法中处理发短信、发邮件、更新库存等
    }

    @Subscribe
    public void change(OrderChangeEvent event) {
        // 处理订单变化后的修改
        // 如发送提醒、更新物流等
    }
}

发送事件

@Service
public class OrderService {

    @Autowired
    private EventBusCenter eventBusCenter;

    public void createOrder() {
        // 处理创建订单
        // ...
        // 发送异步事件
        eventBusCenter.postAsync(new OrderCreatedEvent(1L, 1L));
    }
}

运行程序

    /**
     * 事件总线测试
     */
    @Test
    void bus() throws Exception {
        orderService.createOrder();
    }

总结

EventBus 的好处在于，它将发生事件的代码和事件处理的代码进行了解耦。

比如系统中很多地方都会修改订单，用户可以自己修改、客服也可以修改、甚至可能是团购没成团系统进行的订单修改，所有这些触发订单修改的地方都要发短信、发邮件，假设以后还要增加其他操作，那么需要修改的地方就比较多。

而如果采用事件驱动的话，只要这些地方抛出事件就可以了，后续的维护是比较简单的。

而且，EventBus 支持同步事件和异步事件，可以满足我们不同场景下的需求。比如发短信，系统完全没必要等在那边，完全是可以异步做的。