说下你可能没用过的EventBus

简介: 一般情况下,我们会做成异步的方式,使用MQ自己发送自己消费,或者说一个线程池搞定,这样的话不影响主业务逻辑,可以提高性能,并且代码做到了解耦。

最近在Code Review的时候发现了这样一个业务场景,某个业务处理完成之后需要通知审核人员,通知的方式包含短信和邮件,所以代码大致是这样:

//业务校验
validate();
//处理业务逻辑
doBusiness();
//发送邮件或者发送其他类型消息
sendMsg(); 

这个对不对呢?

基于这种普遍的业务场景来说,一般首先我们会考虑同步或者异步发送的问题。

同步的话对接口RT有影响,而且和业务逻辑耦合在一起,这样的做法肯定不太好。

一般情况下,我们会做成异步的方式,使用MQ自己发送自己消费,或者说一个线程池搞定,这样的话不影响主业务逻辑,可以提高性能,并且代码做到了解耦。

然后还有就是数据一致性的问题,邮件一定要发送成功吗?

大多数时候其实我们并不要求邮件一定100%发送成功,失败了就失败好了,监控告警打点做好失败率不要超过阈值就好,还有就是消息服务一旦收到请求应该自己保证消息能够投递。

所以总的来说,使用MQ发送消息自己消费处理,或者线程池异步处理,最后自己搞个补偿的逻辑就能处理好这类问题。

那么,今天要说的是这两个解决方案之外的处理方式,对于这种场景其实我们可以用EventBus来解决。

EventBus使用

看名字就知道,EventBus是事件总线的意思,它是Google Guava库的一个工具,基于观察者模式可以做到进程内的代码解耦作用。

就拿上面的例子来说,引入一个MQ太重了,其实不太需要这样做,EventBus也能达到这个效果,和MQ相比他只能提供进程内的消息事件传递,这对于我们这种业务场景来说足够了不是吗?

我们先看EventBus怎么来使用,一般先创建一个EventBus实例。

//1.创建EventBus
private static EventBus eventBus = new EventBus();

第二步,创建一个事件消息订阅者,处理方式非常简单,只要在我们希望去处理事件的方法上加上@Subscribe注解即可。

形参只能有一个,如果定义0个或者多个的话运行就会报错。

public class EmailMsgHandler {
    @Subscribe
    public void handle(Long businessId) {
        System.out.println("send email msg" + businessId);
    }
}

第三步,注册事件。

eventBus.register(new EmailMsgHandler());

第四步,发送事件。

eventBus.post(1L);

这就是一个EventBus使用的最简单例子,下面我们看看结合开头说的例子怎么处理。

结合实际

比如上面说的案例,举例来说比如注册和用户下单的场景,都需要发送消息和邮件给用户。

EventBus并不强制说我们一定要用单例模式,因为他的创建销毁成本比较低,所以更多是根据我们的业务场景和上下文自己来选择。

public class UserService {
    private static EventBus eventBus = new EventBus();
    public void regist(){
        Long userId = 1L;
        eventBus.register(new EmailMsgHandler());
        eventBus.register(new SmsMsgHandler());
        eventBus.post(userId);
    }
}
public class BookingService {
    private static EventBus eventBus = new EventBus();
    public void booking(){
        //业务逻辑
        Long bookingId = 2L;
        eventBus.register(new EmailMsgHandler());
        eventBus.register(new SmsMsgHandler());
        eventBus.post(bookingId);
    }
}

然后在业务逻辑处理完成之后,分别去注册了邮件和短信两个事件订阅者。

public class EmailMsgHandler {
    @Subscribe
    public void handle(Long businessId) {
        System.out.println("send email msg" + businessId);
    }
}
public class SmsMsgHandler {
    @Subscribe
    public void handle(Long businessId) {
        System.out.println("send sms msg" + businessId);
    }
}

最后我们发送事件,用户注册我们发送了一个用户ID,下单成功我们发送了一个订单ID。

再写一个测试类去测试一下,分别创建两个service,然后分别调用方法。

public class EventBusTest {
    public static void main(String[] args) {
        UserService userService = new UserService();
        userService.regist();
        BookingService bookingService = new BookingService();
        bookingService.booking();
    }
}

执行测试类,我们可以看到输出,分别去执行了我们的事件订阅的方法。

send email msg1
send sms msg1
send email msg2
send sms msg2

使用起来你会发现非常简单,对于希望轻量级简单地做到解耦使用EventBus非常合适。

注意别踩坑

首先,注意一下例子中的参数都是Long类型,如果事件的参数是其他类型的话,那么消息是无法接受到的,比如我们把下单中发送的订单ID改成String类型然后会发现没有消费了,因为我们没有定义一个参数类型是String的方法。

public class BookingService {
    private static AsyncEventBus eventBus = new AsyncEventBus(Executors.newFixedThreadPool(3));
    public void booking(){
        //业务逻辑
        String bookingId = "2";
        eventBus.register(new EmailMsgHandler());
        eventBus.register(new SmsMsgHandler());
        eventBus.post(bookingId);
    }
}
//输出
send email msg1
send sms msg1

EmailMsgHandlerSmsMsgHandler都新增一个接收String类型的订阅方法,这样就可以接收到了。

@Subscribe
public void handle(String businessId) {
 System.out.println("send email msg for string" + businessId);
}
@Subscribe
public void handle(String businessId) {
 System.out.println("send sms msg for string" + businessId);
}
//输出
send sms msg1
send email msg1
send email msg for string2
send sms msg for string2

除此之外,其实我们可以定义一个DeadEvent来处理这种情况,它相当于是一个默认的处理方式,当没有匹配的事件类型参数的话就会默认发送一个DeadEvent事件。

定义一个默认处理器。

public class DefaultEventHandler {
    @Subscribe
    public void handle(DeadEvent event) {
        System.out.println("no subscriber," + event);
    }
}

BookingService新增一个pay()支付方法,下单完了去支付,注册我们的默认事件。

public void pay(){
  //业务逻辑
  eventBus.register(new DefaultEventHandler());
  eventBus.post(new Payment(UUID.randomUUID().toString()));
}
@ToString
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Payment {
    private String paymentId;
}

执行测试bookingService.pay()看到输出结果:

no subscriber,DeadEvent{source=AsyncEventBus{default}, event=Payment(paymentId=255da942-7128-4bd1-baca-f0a8e569ed88)}

源码分析

OK,简单的介绍就到这里,那其实到目前为止我们说的这个都是同步调用的,这不太符合我们的要求,我们当然使用异步处理更好。

那就看看源码它是怎么实现的。

@Beta
public class EventBus {
  private static final Logger logger = Logger.getLogger(EventBus.class.getName());
  private final String identifier;
  private final Executor executor;
  private final SubscriberExceptionHandler exceptionHandler;
  private final SubscriberRegistry subscribers = new SubscriberRegistry(this);
  private final Dispatcher dispatcher;
  public EventBus() {
    this("default");
  }
  public EventBus(String identifier) {
    this(
        identifier,
        MoreExecutors.directExecutor(),
        Dispatcher.perThreadDispatchQueue(),
        LoggingHandler.INSTANCE);
  }
}

identifier就是个名字,标记,默认就是default

executor执行器,默认创建一个MoreExecutors.directExecutor(),事件订阅者根据你自己提供的executor来决定如何执行事件订阅的处理方式。

exceptionHandler是异常处理器,默认创建的就是打点日志。

subscribers就是我们的消费者,订阅者。

dispatcher用来做事件分发。

默认创建的executor是一个MoreExecutors.directExecutor(),看到command.run()你就会发现他这不就是同步执行嘛。

public static Executor directExecutor() {
 return DirectExecutor.INSTANCE;
}
private enum DirectExecutor implements Executor {
 INSTANCE;
@Override
public void execute(Runnable command) {
 command.run();
}
@Override
public String toString() {
 return "MoreExecutors.directExecutor()";
}

同步执行还是不太好,我们希望不光给我们解耦,还要异步执行,EventBus给我们提供了AsyncEventBusExecutor我们自己传入就好了。

public class AsyncEventBus extends EventBus {
  public AsyncEventBus(String identifier, Executor executor) {
    super(identifier, executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
  }
  public AsyncEventBus(Executor executor, SubscriberExceptionHandler subscriberExceptionHandler) {
    super("default", executor, Dispatcher.legacyAsync(), subscriberExceptionHandler);
  }
  public AsyncEventBus(Executor executor) {
    super("default", executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
  }

上面的代码我们改成异步的,这样不就好起来了嘛,这样的话,实际上可以结合我们自己的线程池来处理了。

private static AsyncEventBus eventBus = new AsyncEventBus(Executors.newFixedThreadPool(3));

OK,这个说清楚了,我们可以顺便再看看事件分发的处理,看到DeadEvent了吧,没有当前事件的订阅者,就会发送一个DeadEvent事件,bingo!

ublic void post(Object event) {
  Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);
   if (eventSubscribers.hasNext()) {
   dispatcher.dispatch(event, eventSubscribers);
  } else if (!(event instanceof DeadEvent)) {
    // the event had no subscribers and was not itself a DeadEvent
    post(new DeadEvent(this, event));
  }
}

总结

OK,这个使用和源码还是比较简单的哈,有兴趣的同学可以自己去瞅瞅,花不了多少工夫。

总的来说,EventBus就是提供了我们一个更优雅的代码解耦的方式,实际工作中的业务你肯定能用上它!

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
安全
自写一个EventBus
entBus,什么是EventBus。 EventBus是事件发布-订阅总线,简单来说监听一个事件,一个方法订阅这个事件,如果事件调用,那么订阅了这个事件的方法也会跟着调用,这就是EventBus。
91 0
|
前端开发
前端学习笔记202303学习笔记第四天-Vue3.0-事件对象event
前端学习笔记202303学习笔记第四天-Vue3.0-事件对象event
89 0
探秘 RxJS Observable 为什么要长成这个样子?!
我们都知道 RxJS Observable 最基础的使用方法:是建立 Observable,即调用 .create API
|
Android开发 索引
EventBus封装到项目架构|青训营笔记
封装该库到自己的项目的目的有两个 便捷绑定和解绑 EventBus 便捷通过 EventBus 发送消息和处理消息 代码美观
EventBus封装到项目架构|青训营笔记
|
缓存 Java Android开发
EventBus原理解析笔记以及案例实战(结合demo)
EventBus原理解析笔记以及案例实战(结合demo)
|
JavaScript
手写代码:实现一个EventBus
EventBus,事件总线。总线一词来自于《计算机组成原理》中的”系统总线“,是指用于连接多个部件的信息传输线,各部件共享的传输介质。我们通常把事件总线也成为自定义事件,一般包含`on`、`once`、`emit`、`off`等方法。在Vue2中想要实现EventBus比较简单,直接暴露出一个`new Vue()`实例即可,以此为思路,我们应该如何自定义实现EventBus呢?
530 0
手写代码:实现一个EventBus
DHL
|
Java API Android开发
EventBus3.1用法详解
EventBus是Android和Java的发布/订阅事件总线。从EventBus3.1开始支持普通Java(非android)项目。GitHub的地址
DHL
252 0
EventBus3.1用法详解
|
存储 缓存 调度
EventBus源码解析
EventBus源码解析
750 0
EventBus源码解析
EventBus简单使用
EventBus简单使用
188 0
|
前端开发 JavaScript
ReactJS实战之事件处理
ReactJS实战之事件处理
120 0
ReactJS实战之事件处理