观察者模式及Guava EventBus

编写不易，转载请注明（http://shihlei.iteye.com/blog/2426888）！

 

一 概述

最近看RxJava，其骨架就是使用的观察者模式，所以对观察者模式做个总结，顺便记录一下Guava EventBus的实现（事件监听，其实也相当于观察者模式）

 

二 观察者模式

1）概述

 

观察者模式：行为模式，提供一种一对多关系绑定对象的方法，一个对象状态需发生改变，绑定对象能收到通知完成自己的业务更新。

 

主要成员：

被观察者（Observerable）：状态变化，通知所有的观察者。

观察者（observer）：接收到“被观察者”的状态变化通知，执行自己的业务。

 

使用场景：

“被观察者”的状态变化需要通知多个“观察者”，但“被观察者”不需要知道观察者的个数具体细节，他们之间互相独立。一般“观察者”之间相互独立，不会彼此影响。

 

2）Demo

 

package x.rx.observer;

import java.util.LinkedList;
import java.util.List;

/**
 * 观察者模式：
 * 1）被观察者（Observerable）：维护一个观察者队列，事件产生时，回调所以的观察者
 * 2）观察者（Observer）：注册到"被观察者"中，接收通知，执行自己的业务
 *
 * @author shilei
 */
public class ObserverPatternDemo {

    public static void main(String[] args) {
        Observerable observerable = new Observerable();
        observerable.register(() -> {
            System.out.println("observer1 handle finish！");
        }).register(() -> {
            System.out.println("observer2 handle finish！");
        });

        observerable.generateEvent();
    }


    /**
     * 观察者
     */
    interface Observer {
        void doEvent();
    }

    /**
     * 被观察者
     */
    static class Observerable {

        private List<Observer> observerList = new LinkedList<>();

        /**
         * 注册观察者
         *
         * @param observer 观察者
         */
        Observerable register(Observer observer) {
            observerList.add(observer);
            return this;
        }

        /**
         * 取消注册
         */
        Observerable unRegister(Observer observer) {
            observerList.remove(observer);
            return this;
        }

        /**
         * 产生事件
         */
        void generateEvent() {
            for (Observer observer : observerList) {
                observer.doEvent();
            }
        }
    }
}

 

三 Guava EventBus

1) 概述

 

Guava EventBus 实现了事件监听器模式，主要提供一套基于注解的事件总线，可以灵活的使用。

 

使用方式：

1）定义一个关注的Event

2）定义一个监听处理方法，@Subscribe 标记，里面实现业务逻辑

3）注册到EventBus即可

 

2）Demo

（1）依赖

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

 

 （2）代码

package x.rx.eventbus;


import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;

/**
 * guava eventbus
 *
 * @author shilei
 */
public class EventBusDemo {

    public static void main(String[] args) {
        EventBus eventBus = new EventBus("demo");
        // 注册
        eventBus.register(new EventListener());
        // 产生事件
        eventBus.post(new Event());
    }

    /**
     * 事件
     */
    static class Event {

    }

    /**
     * 事件监听器
     */
    static class EventListener {

        @Subscribe //这里标记这个方法是事件处理方法
        public void handle1(Event event) {
            System.out.println("handle1 finish! ");
        }

        @Subscribe
        public void handle2(Event event) {
            System.out.println("handle2 finish! ");
        }

    }
}

 

3）源码分析

 

（1）事件注册到总线：EventBus的register()

 

private final SubscriberRegistry subscribers = new SubscriberRegistry(this);
 
  /**
   * Registers all subscriber methods on {@code object} to receive events.
   *
   * @param object object whose subscriber methods should be registered.
   */
public void register(Object object) {
    subscribers.register(object);
}

 

 “被观察者”的核心，提供一个队列，维护所有需要通知的观察者，Guava EventBus 这个工作代理给了 SubscriberRegistry ， SubscriberRegistry 提供任何类型的事件到事件处理类的绑定关系。

 

看看注册的细节：

 

 /**
   * All registered subscribers, indexed by event type.
   *
   * <p>The {@link CopyOnWriteArraySet} values make it easy and relatively lightweight to get an
   * immutable snapshot of all current subscribers to an event without any locking.
   */
  private final ConcurrentMap<Class<?>, CopyOnWriteArraySet<Subscriber>> subscribers =
      Maps.newConcurrentMap();

  /**
   * Registers all subscriber methods on the given listener object.
   */
  void register(Object listener) {
    Multimap<Class<?>, Subscriber> listenerMethods = findAllSubscribers(listener);

    for (Map.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<Subscriber>();
        eventSubscribers =
            MoreObjects.firstNonNull(subscribers.putIfAbsent(eventType, newSet), newSet);
      }

      eventSubscribers.addAll(eventMethodsInListener);
    }
  }

 

 

    内部是一个ConcrurentMap，key 是事件class，value是个 CopyOnWriteArraySet<Subscriber> 用于通知，最重要的方法  findAllSubscribers(listener)  ，他会通过反射，找到 @Subscribe 注解的方法，并关联他绑定的事件。获取到就添加到 subscribers（类型ConcrurentMap）中，这个subscribers 维护了所有的事件和事件处理器的绑定关系。

 

    看看findAllSubscribers(listener) 的细节：

 

  /**
   * Returns all subscribers for the given listener grouped by the type of event they subscribe to.
   */
  private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
    Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();
    Class<?> clazz = listener.getClass();
    for (Method method : getAnnotatedMethods(clazz)) {
      Class<?>[] parameterTypes = method.getParameterTypes();
      Class<?> eventType = parameterTypes[0];
      methodsInListener.put(eventType, Subscriber.create(bus, listener, method));
    }
    return methodsInListener;
  }

   

    很常规的反射和Annotation处理，不过事件一定是处理方法的第一个参数。

    

    就此，EventBus就有了要通知的订阅者列表了

 

（2）提交事件，调用订阅者方法

 

这里就很简单了，根据事件类型，从 subscribers（类型ConcrurentMap）获得相应的订阅者集合，通过反射调用下方法就ok

  

  /**
   * Posts an event to all registered subscribers. This method will return successfully after the
   * event has been posted to all subscribers, and regardless of any exceptions thrown by
   * subscribers.
   *
   * <p>If no subscribers have been subscribed for {@code event}'s class, and {@code event} is not
   * already a {@link DeadEvent}, it will be wrapped in a DeadEvent and reposted.
   *
   * @param event event to post.
   */
  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));
    }
  }

  

（3）总结：

 

EventBus 在观察者模式上做了通用性的抽象，可以定义任何事件，和基于注解的事件处理器，还是非常有用的。