Google Guava EventBus 消息发布-订阅异步调用使用

比眉伴天荒 2023-07-03 12:05 51阅读 0赞

`EventBus` **是Google.Guava提供的消息发布-订阅类库，它实现了观察者设计模式，消息通知负责人通过EventBus去注册/注销观察者，最后由消息通知负责人给观察者发布消息**。

前提：在pom.xml中引入guava包

<dependency>  
                <groupId>com.google.guava</groupId>  
                <artifactId>guava</artifactId>  
                <version>19.0</version>  
            </dependency>

**demo**

（1）EventBusCenter.java

package com.lance.google.event.bus;  
      
    import com.google.common.eventbus.EventBus;  
      
    /** 
     *  
     */  
    public class EventBusCenter {  
      
        private static EventBus eventBus = new EventBus();  
      
        private EventBusCenter() {  
      
        }  
      
        public static EventBus getInstance() {  
            return eventBus;  
        }  
      
        public static void register(Object obj) {  
            eventBus.register(obj);  
        }  
      
        public static void unregister(Object obj) {  
            eventBus.unregister(obj);  
        }  
      
        public static void post(Object obj) {  
            eventBus.post(obj);  
        }  
     
    }

(2) 观察者1

package com.lance.google.event.bus;  
      
    import com.google.common.eventbus.Subscribe;  
      
    /** 
     *  
     */  
    public class DataObserver1 {  
      
        /** 
         * 只有通过@Subscribe注解的方法才会被注册进EventBus 
         * 而且方法有且只能有1个参数 
         * 
         * @param msg 
         */  
        @Subscribe  
        public void func(String msg) {  
            System.out.println("String msg: " + msg);  
        }  
      
    }

(3) 观察者2

package com.lance.google.event.bus;  
      
    import com.google.common.eventbus.Subscribe;  
      
    /** 
     *  
     */  
    public class DataObserver2 {  
        /** 
         * post() 不支持自动装箱功能,只能使用Integer,不能使用int,否则handlersByType的Class会是int而不是Intege 
         * 而传入的int msg参数在post(int msg)的时候会被包装成Integer,导致无法匹配到 
         */  
        @Subscribe  
        public void func(Integer msg) {  
            System.out.println("Integer msg: " + msg);  
        }  
    }

(4) Test.java

package com.lance.google.event.bus;  
      
    /** 
     *  
     */  
    public class Test {  
      
        public static void main(String[] args) throws InterruptedException {  
      
            DataObserver1 observer1 = new DataObserver1();  
            DataObserver2 observer2 = new DataObserver2();  
      
            EventBusCenter.register(observer1);  
            EventBusCenter.register(observer2);  
      
            System.out.println("============   start  ====================");  
      
            // 只有注册的参数类型为String的方法会被调用  
            EventBusCenter.post("post string method");  
            EventBusCenter.post(123);  
      
            System.out.println("============ after unregister ============");  
            // 注销observer2  
            EventBusCenter.unregister(observer2);  
            EventBusCenter.post("post string method");  
            EventBusCenter.post(123);  
      
            System.out.println("============    end           =============");  
        }  
    }

--------------------

输出结果:

String msg: post string method  
    Integer msg: 123  
    ============ after unregister ============  
    String msg: post string method  
    ============    end           =============

*  EventBus的使用注意问题：

1.  代码可读性很差，项目中使用的时候，从post的地方，查询handle使用，都是使用ide的搜索服务，问题很难定位，不如普通的接口调用方便查询；
2.  由于EventBus是将消息队列放入到内存中的，listener消费这个消息队列，故系统重启之后，保存或者堆积在队列中的消息丢失。

====================================

**同步模式  
观察者类**

/**
     * 用于响应事件
     *
     * @author  
     * @version 2019/2/7 21:21
     */
    public class MyObserver {
    
        // 添加该注解即可
        @Subscribe
        public void onEvent(Object obj) {
            if (obj != null) {
                System.out.println("obj = " + obj);
            }
        }
    
    }

**EventBus使用**

// 实例化EventBus对象
    // 定义一个简称Test，用于日志
    EventBus eventBus = new EventBus("Test");
    // 实例化观察者对象
    MyObserver observer = new MyObserver();
    // 订阅，即注册观察者
    eventBus.register(observer);
    // 分发，即触发事件
    eventBus.post("Hello World!");
    
    // 注销
    eventBus.unregister(observer);

**异步模式  
观察者类**

**EventBus使用**

// 构建一个 Executor 用于异步执行
    Executor executor = new Executor() {
        @Override
        public void execute(Runnable command) {
            new Thread(command).start();
        }
    };
    // 实例化AsyncEventBus对象
    AsyncEventBus asyncEventBus = new AsyncEventBus("Async", executor);
    // 其他的使用和同步一样
    asyncEventBus.register(observer);
    asyncEventBus.post("Async Hello World!");
    asyncEventBus.unregister(observer);

**注意**  
需要防止并发调用时，可在@Subscribe注解下再加上@AllowConcurrentEvents  
EventBus并未直接实现单例，可以根据自己的业务来随机应对  
Subscribe的方法必须是Public的

**==================详解=================**

## １、如何使用 ##

*List-1.1*

import com.google.common.eventbus.AllowConcurrentEvents;
    import com.google.common.eventbus.AsyncEventBus;
    import com.google.common.eventbus.Subscribe;
    import org.junit.Before;
    import org.junit.Test;
    
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    public class EventBusTest {
        private AsyncEventBus asyncEventBus;
    
        @Before
        public void before(){
            asyncEventBus=new AsyncEventBus(Executors.newFixedThreadPool(3));
            asyncEventBus.register(this);
        }
    
        @Subscribe
        @AllowConcurrentEvents
        public void subscribe(Object object){
            System.out.println("收到:"+object);
        }
    
        @Test
        public void test_sendMsg() throws InterruptedException {
            System.out.println("开始发送消息");
            asyncEventBus.post("这是消息");
            System.out.println("开始睡眠");
            TimeUnit.SECONDS.sleep(5L);
        }
    }

List-1.1中，方法subscribe是接收者，方法test\_sendMsg中post消息后，方法subscribe就会收到消息。这是因为方法subscribe上有注解Subscribe。

为什么要在方法subscribe上加上注解AllowConcurrentEvents，加上这个才能达到真正的异步，这要看底层源码，下面我们会来分析。

## ２、register方法底层实现 ##

AsyncEventBus的构造方法如下List-2.1所示

*List-2.1*

public AsyncEventBus(Executor executor) {
      super("default", executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
    }

来看下register方法的实现，如下图2.1所示，步骤３中，会找到所有方法上有Subscribe注解的方法，在步骤６中，会判断方法上是否注解AllowConcurrentEvents，如果有，则返回Subscriber，如果没有则返回SynchronizedSubscriber。

![format_png][]

图2.1 AsyncEventBus的register方法

SynchronizedSubscriber和Subscriber的区别如下，SynchronizedSubscriber重复了父类的invokeSubscriberMethod，并加上了锁关键字synchronized，所以List-1.1中的方法上如果没有注解AllowConcurrentEvents，那么是不会真正的并发的，我看了网上的例子，很多描述的不全面。

List-2.2

@VisibleForTesting
    static final class SynchronizedSubscriber extends Subscriber {
    
      private SynchronizedSubscriber(EventBus bus, Object target, Method method) {
        super(bus, target, method);
      }
    
      @Override
      void invokeSubscriberMethod(Object event) throws InvocationTargetException {
        synchronized (this) {
          super.invokeSubscriberMethod(event);
        }
      }
    }

## ３、post方法底层实现 ##

![format_png 1][]

图3.1 AsyncEventBus的post实现

步骤6、７的代码如下List-3.1，可以看到List-2.2中涉及的invokeSubscriberMethod在这里使用。

*List-3.1*

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

guava提供了三个Dispatcher，上面使用了LegacyAsyncDispatcher，LegacyAsyncDispatcher的dispatch实现如下List-3.2所示，可以看到，是将event和subscriber放入到ConcurrentLinkedQueue中，之后再从queue中poll出来，再调用subscribe的dispatchEvent方法。为什么先放到queue中，之后在poll出来，这是有考虑的，是为了应用整体的吞吐量考虑。

*List-3.2*

private final ConcurrentLinkedQueue<EventWithSubscriber> queue =
        Queues.newConcurrentLinkedQueue();
    
    @Override
    void dispatch(Object event, Iterator<Subscriber> subscribers) {
      checkNotNull(event);
      while (subscribers.hasNext()) {
        queue.add(new EventWithSubscriber(event, subscribers.next()));
      }
    
      EventWithSubscriber e;
      while ((e = queue.poll()) != null) {
        e.subscriber.dispatchEvent(e.event);
      }
    }

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9hc2sucWNsb3VkaW1nLmNvbS9odHRwLXNhdmUveWVoZS01MDg5Nzc0L2I0MWIwb3dia24uanBlZw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9hc2sucWNsb3VkaW1nLmNvbS9odHRwLXNhdmUveWVoZS01MDg5Nzc0L3VtNW1qdHNpdW8uanBlZw?x-oss-process=image/format,png