观察者模式及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;
}
就此，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 在观察者模式上做了通用性的抽象，可以定义任何事件，和基于注解的事件处理器，还是非常有用的。
上一篇： 响应式编程 RxJava
下一篇： mysql启动流程