Guava之eventBus异步事件总线的使用及源码分析

克虏伯

发布于 2019-04-15 10:14:18

2.5K0

发布于 2019-04-15 10:14:18

文章被收录于专栏：软件开发-青出于蓝

最近使用guava的eventBus，记录下。

１、如何使用

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。

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

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