文档章节

EventBus原理深度解析

yangjianzhou
 yangjianzhou
发布于 09/26 01:29
字数 1358
阅读 2916
收藏 93

一、问题描述

在工作中,经常会遇见使用异步的方式来发送事件,或者触发另外一个动作:经常用到的框架是MQ(分布式方式通知)。如果是同一个jvm里面通知的话,就可以使用EventBus。由于EventBus使用起来简单、便捷,因此,工作中会经常用到。深入理解该框架的原理就很有必要。

二、框架解析

2.1、组织结构

eventbus的组织结构如下:

eventbus主要有以下几部分组成:

1、eventbus、asyncEventBus:事件发送器。

2、event:事件承载单元。

3、SubscriberRegistry:订阅者注册器,将订阅者注册到event上,即将有注解Subscribe的方法和event绑定起来。

4、Dispatcher:事件分发器,将事件的订阅者调用来执行。

5、Subscriber、SynchronizedSubscriber:订阅者,并发订阅还是同步订阅。

2.2、运行原理

1、eventbus是基于注册监听的方式来运行的,因此,首先需要将eventbus,然后才会有事件及监听者。新建eventbus或者AsyncEventBus的方式如下:

 EventBus eventBus = new EventBus();

或者

  BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(20);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20,
                30, TimeUnit.SECONDS, workQueue);
        AsyncEventBus asyncEventBus = new AsyncEventBus(executor);

2、注册监听者。

eventBus.register(eventListener);

底层就是将类eventListener中所有注解有Subscribe的方法与其Event对放在一个map中(一个event可以对应多个Subscribe的方法)。实现如下:

  void register(Object listener) {
    Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);

    for (Entry<Class<?>, Collection<Subscriber>> entry : listenerMethods.asMap().entrySet()) {
      Class<?> eventType = entry.getKey();
      Collection<Subscriber> eventMethodsInListener = entry.getValue();

      CopyOnWriteArraySet<Subscriber> eventSubscribers = subscribers.get(eventType);

      if (eventSubscribers == null) {
        CopyOnWriteArraySet<Subscriber> newSet = new CopyOnWriteArraySet<>();
        eventSubscribers =
            MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);
      }

      eventSubscribers.addAll(eventMethodsInListener);
    }
  }

3、事件发送:执行指定事件类型的订阅者(包含了method),从订阅者中获取指定事件的订阅者,然后按照规则(同步、异步)执行指定的方法。

  public 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));
    }
  }

上述代码说明,如果事件没有监听者,就当作死亡事件来对待。

 /** Dispatches {@code event} to this subscriber using the proper executor. */
  final void dispatchEvent(final Object event) {
    executor.execute(
        new Runnable() {
          @Override
          public void run() {
            try {
              invokeSubscriberMethod(event);
            } catch (InvocationTargetException e) {
              bus.handleSubscriberException(e.getCause(), context(event));
            }
          }
        });
  }
  void invokeSubscriberMethod(Object event) throws InvocationTargetException {
    try {
      method.invoke(target, checkNotNull(event));
    } catch (IllegalArgumentException e) {
      throw new Error("Method rejected target/argument: " + event, e);
    } catch (IllegalAccessException e) {
      throw new Error("Method became inaccessible: " + event, e);
    } catch (InvocationTargetException e) {
      if (e.getCause() instanceof Error) {
        throw (Error) e.getCause();
      }
      throw e;
    }
  }

这里就说明,最后就是被订阅的方法被调用。

4、EventBus与AsyncEventBus的区别

从字面上看,AsyncEventBus是异步的EventBus,那么EventBus应该就是同步的了。EventBus的executor为MoreExecutors.directExecutor(),其实现如下:

  public static Executor directExecutor() {
    return DirectExecutor.INSTANCE;
  }

  /** See {@link #directExecutor} for behavioral notes. */
  private enum DirectExecutor implements Executor {
    INSTANCE;

    @Override
    public void execute(Runnable command) {
      command.run();
    }

    @Override
    public String toString() {
      return "MoreExecutors.directExecutor()";
    }
  }

其execute方法直接执行线程的run方法,即同步调用run方法执行。EventBus的dispatcher为PerThreadQueuedDispatcher。其dispatch方法如下:

 @Override
    void dispatch(Object event, Iterator<Subscriber> subscribers) {
      checkNotNull(event);
      checkNotNull(subscribers);
      Queue<Event> queueForThread = queue.get();
      queueForThread.offer(new Event(event, subscribers));

      if (!dispatching.get()) {
        dispatching.set(true);
        try {
          Event nextEvent;
          while ((nextEvent = queueForThread.poll()) != null) {
            while (nextEvent.subscribers.hasNext()) {
              nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
            }
          }
        } finally {
          dispatching.remove();
          queue.remove();
        }
      }
    }

dispatchEvent的实现如下:

  final void dispatchEvent(final Object event) {
    executor.execute(
        new Runnable() {
          @Override
          public void run() {
            try {
              invokeSubscriberMethod(event);
            } catch (InvocationTargetException e) {
              bus.handleSubscriberException(e.getCause(), context(event));
            }
          }
        });
  }

因此,整个执行过程如下:

整个过程都是同步方式执行,因此,EventBus是同步的。

AsyncEventBus的dispatcher为LegacyAsyncDispatcher,executor为自己指定的线程池。运行流程如下:

虚线为线程池异步调度,因此,AsyncEventBus为异步方式。

5、AllowConcurrentEvents的作用

它所在的代码为:

  static Subscriber create(EventBus bus, Object listener, Method method) {
    return isDeclaredThreadSafe(method)
        ? new Subscriber(bus, listener, method)
        : new SynchronizedSubscriber(bus, listener, method);
  }

  private static boolean isDeclaredThreadSafe(Method method) {
    return method.getAnnotation(AllowConcurrentEvents.class) != null;
  }

即如果订阅者方法上有注解AllowConcurrentEvents,则返回Subscriber,否则,返回SynchronizedSubscriber。SynchronizedSubscriber的字面意思为同步订阅者,它的实现代码为:

   @Override
    void invokeSubscriberMethod(Object event) throws InvocationTargetException {
      synchronized (this) {
        super.invokeSubscriberMethod(event);
      }
    }

即没有使用注解AllowConcurrentEvents的订阅者,在并发环境中,都是串行执行。这在高并发环境中,会严重影响性能。

三、使用案例

3.1、eventbus定义

@Configuration
public class ConfigBean {

    @Bean
    public EventBus executorService() {
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(20);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20,
                30, TimeUnit.SECONDS, workQueue);
        return new AsyncEventBus(executor);
    }
}

3.2、注册与事件发送

@Service
public class TestService implements InitializingBean {

    @Autowired
    private EventListener eventListener ;

    @Autowired
    private EventBus eventBus ;

    public void postEvent(){
        eventBus.post(new LoginEvent("iwill","123456"));
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        eventBus.register(eventListener);
    }
}

3.3、订阅者定义

package com.iwill.eventBus.listener;

import com.google.common.eventbus.Subscribe;
import com.iwill.eventBus.event.LoginEvent;
import com.iwill.eventBus.event.RegisterEvent;
import org.springframework.stereotype.Component;

@Component
public class EventListener {

    @Subscribe
    public void subscribeLoginEvent1(LoginEvent event){
        System.out.println("method 1 : receive login event ");
    }

    @Subscribe
    public void subscribeLoginEvent2(LoginEvent event){
        System.out.println("method 2 : receive login event ");
    }

    @Subscribe
    public void subscribeRegisterEvent(RegisterEvent event){
        try{
            Thread.sleep(10000L);
        }catch (Exception exp){
            exp.printStackTrace();
        }
        System.out.println("method  : receive register event ");
    }
}

四、注意事项

1、在高并发的环境下使用AsyncEventBus时,发送事件可能会出现异常,因为它使用的线程池,当线程池的线程不够用时,会拒绝接收任务,就会执行线程池的拒绝策略,如果需要关注是否提交事件成功,就需要将线程池的拒绝策略设为抛出异常,并且try-catch来捕获异常。如下:

     try {
            eventBus.post(new LoginEvent("iwill", "123456"));
        }catch (Exception exp){
            //TODO 落表或者其他处理
        }

2、本文用到的guava版本如下:

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

 

© 著作权归作者所有

共有 人打赏支持
yangjianzhou
粉丝 12
博文 10
码字总数 16049
作品 0
闵行
程序员
加载中

评论(2)

久永
久永
这个说的就是 事件聚集器 吧?我在.net中自己实现了一个。
z201
z201
有必要mark一下。
guava eventbus源码解析

说在前面 本文转自“天河聊技术”微信公众号 事件驱动模型设计是一种优雅的程序设计方式,实现有很多,原理都是发布与订阅,观察者设计模式实现,java自带的实现、spring ioc的事件驱动模型,...

天河2018
07/07
0
0
Guava库学习:学习Guava EventBus(一)EventBus

在软件开发过程中,对象信息的分享以及相互直接的协作是必须的,困难在于确保对象之间的沟通是有效完成的,而不是拥有成本高度耦合的组件。当对象对其他组件的责任有太多的细节时,它被认为是...

Realfighter
2014/12/29
0
0
为JFinal添加event消息事件

在之前使用spring mvc的时候,在复杂的下单和支付中有一部分功能使用的Spring事件驱动模型去完成!具体优点不啰嗦了,发现涛哥有篇文章讲得比较详细:[详解Spring事件驱动模型][1] 最初准备基...

如梦技术
2015/04/28
0
20
Android EventBus库的使用

在编程过程中,当我们想通知其他组件某些事情发生时,我们通常使用观察者模式,正式因为观察者模式非常常见,所以在jdk1.5中已经帮助我们实现了观察者模式,我们只需要简单的继承一些类就可以...

恰同学少年
2015/08/03
0
0
【Android翻译】组件通信模式

目标:避免紧耦合 本文对原文进行了精简 原文链接:Communication patterns for application components 紧耦合 组件之间相互持有引用,以及直接调用方法.在下面的代码中,MenuFragment持有Magaz...

xesam
2015/04/23
0
0

没有更多内容

加载失败,请刷新页面

加载更多

kubeadm部署kubernetes集群

一、环境要求 这里使用RHEL7.5 master、etcd:192.168.10.101,主机名:master node1:192.168.10.103,主机名:node1 node2:192.168.10.104,主机名:node2 所有机子能基于主机名通信,编辑...

人在艹木中
今天
6
0
Shell特殊符号总结以及cut,sort,wc,uniq,tee,tr,split命令

特殊符号总结一 * 任意个任意字符 ? 任意一个字符 # 注释字符 \ 脱义字符 | 管道符 # #号后的备注被忽略[root@centos01 ~]# ls a.txt # 备注 a.txt[root@centos01 ~]# a=1[root@centos01...

野雪球
今天
2
0
OSChina 周二乱弹 —— 程序员圣衣

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @达尔文:分享Skeeter Davis的单曲《The End of the World》 《The End of the World》- Skeeter Davis 手机党少年们想听歌,请使劲儿戳(这里...

小小编辑
今天
14
0
[ python import module ] 导入模块

import moudle_name ----> import module_name.py ---> import module_name.py文件路径 -----> sys.path (这里进行查找文件) # from app.web import Personimport app.web.Person as Pe......

_______-
昨天
5
0
Redis性能问题排查解决手册

一、性能相关的数据指标 通过Redis-cli命令行界面访问到Redis服务器,然后使用info命令获取所有与Redis服务相关的信息。通过这些信息来分析文章后面提到的一些性能指标。 nfo命令输出的数据可...

IT--小哥
昨天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部