投稿
  1. 极客之音首页
  2. 后端开发
  3. Java

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

小半 • • Java • 阅读 606

导读：本篇文章讲解 Google Guava EventBus 消息发布-订阅异步调用使用，希望对大家有帮助，欢迎收藏，转发！站点地址：www.bmabk.com

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

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

<!-- https://mvnrepository.com/artifact/com.google.guava/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);

异步模式
观察者类
 

/**
 * 用于响应事件
 *
 * @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使用
 

// 构建一个 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。

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

　　　　　　　　　　　　　　　　　　图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方法底层实现

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

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

 

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/13999.html

赞 (0)
小半的头像小半
0 0

相关推荐

极客之音——专业性很强的中文编程技术网站，欢迎收藏到浏览器，订阅我们！