0. 不由自主，恍恍惚惚，又走回头路，再看一眼有过的幸福…
太棒了流沙!!!
0.1 引用
https://blog.csdn.net/majunzhu/article/details/100869562
ApplicationEvent事件机制源码分析
单机环境下优雅地使用事件驱动进行代码解耦
1. JDK
1.1 监听器模式
spring的ApplicationEvent机制也是基于jdk监听器规范实现的
1.1.1 EventListener
标记接口
package java.util;
/**
* A tagging interface that all event listener interfaces must extend.
* @since JDK1.1
*/
public interface EventListener {
}
1.1.2 EventObject
由此可见，jdk的事件规范: 事件本身持有事件源的引用
package java.util;
/**
* <p>
* The root class from which all event state objects shall be derived.
* <p>
* All Events are constructed with a reference to the object, the "source",
* that is logically deemed to be the object upon which the Event in question
* initially occurred upon.
*
* @since JDK1.1
*/
public class EventObject implements java.io.Serializable {
/**
* The object on which the Event initially occurred.
*/
protected transient Object source;
/**
* Constructs a prototypical Event.
*
* @param source The object on which the Event initially occurred.
* @exception IllegalArgumentException if source is null.
*/
public EventObject(Object source) {
if (source == null)
throw new IllegalArgumentException("null source");
this.source = source;
}
}
1.1.3 事件源
- 产生事件的对象,需要自己实现
- 比如说 spring.ApplicationEvent 的实现类就持有 其事件源 applicationContext
其实 事件由事件源产生，但事件本身并没有传播能力，甚至是路由到特定的监听器受处理。这意味着我们可以按照自己的想法大胆去实现!
再借用spring.applicationEvent 举例，applicationContext 可以作为事件源，调用 multicaster(路由的方式可以理解为传播)给所有的applicationListener投递事件
1.2 观察者模式
- 用的很少(jdk的实现挺过时的)，比较理论，某种程度上与监听器很接近，并且扩展性不如监听器模式
- 目前没有找到其实现类
1.2.1 可观测对象
package java.util;
/**
* This class represents an observable object, or "data"
* in the model-view paradigm. It can be subclassed to represent an
* object that the application wants to have observed.
* <p>
* An observable object can have one or more observers. An observer
* may be any object that implements interface <tt>Observer</tt>. After an
* observable instance changes, an application calling the
* <code>Observable</code>'s <code>notifyObservers</code> method
* causes all of its observers to be notified of the change by a call
* to their <code>update</code> method.
* <p>
* The order in which notifications will be delivered is unspecified.
* The default implementation provided in the Observable class will
* notify Observers in the order in which they registered interest, but
* subclasses may change this order, use no guaranteed order, deliver
* notifications on separate threads, or may guarantee that their
* subclass follows this order, as they choose.
* <p>
* Note that this notification mechanism has nothing to do with threads
* and is completely separate from the <tt>wait</tt> and <tt>notify</tt>
* mechanism of class <tt>Object</tt>.
* <p>
* When an observable object is newly created, its set of observers is
* empty. Two observers are considered the same if and only if the
* <tt>equals</tt> method returns true for them.
*
* @author Chris Warth
* @see java.util.Observable#notifyObservers()
* @see java.util.Observable#notifyObservers(java.lang.Object)
* @see java.util.Observer
* @see java.util.Observer#update(java.util.Observable, java.lang.Object)
* @since JDK1.0
*/
public class Observable {
private boolean changed = false;
private Vector<Observer> obs;
/** Construct an Observable with zero Observers. */
public Observable() {
obs = new Vector<>();
}
/**
* Adds an observer to the set of observers for this object, provided
* that it is not the same as some observer already in the set.
* The order in which notifications will be delivered to multiple
* observers is not specified. See the class comment.
*
* @param o an observer to be added.
* @throws NullPointerException if the parameter o is null.
*/
public synchronized void addObserver(Observer o) {
if (o == null)
throw new NullPointerException();
if (!obs.contains(o)) {
obs.addElement(o);
}
}
/**
* Deletes an observer from the set of observers of this object.
* Passing <CODE>null</CODE> to this method will have no effect.
* @param o the observer to be deleted.
*/
public synchronized void deleteObserver(Observer o) {
obs.removeElement(o);
}
/**
* If this object has changed, as indicated by the
* <code>hasChanged</code> method, then notify all of its observers
* and then call the <code>clearChanged</code> method to
* indicate that this object has no longer changed.
* <p>
* Each observer has its <code>update</code> method called with two
* arguments: this observable object and <code>null</code>. In other
* words, this method is equivalent to:
* <blockquote><tt>
* notifyObservers(null)</tt></blockquote>
*
* @see java.util.Observable#clearChanged()
* @see java.util.Observable#hasChanged()
* @see java.util.Observer#update(java.util.Observable, java.lang.Object)
*/
public void notifyObservers() {
notifyObservers(null);
}
/**
* If this object has changed, as indicated by the
* <code>hasChanged</code> method, then notify all of its observers
* and then call the <code>clearChanged</code> method to indicate
* that this object has no longer changed.
* <p>
* Each observer has its <code>update</code> method called with two
* arguments: this observable object and the <code>arg</code> argument.
*
* @param arg any object.
* @see java.util.Observable#clearChanged()
* @see java.util.Observable#hasChanged()
* @see java.util.Observer#update(java.util.Observable, java.lang.Object)
*/
public void notifyObservers(Object arg) {
/*
* a temporary array buffer, used as a snapshot of the state of
* current Observers.
*/
Object[] arrLocal;
synchronized (this) {
/* We don't want the Observer doing callbacks into
* arbitrary code while holding its own Monitor.
* The code where we extract each Observable from
* the Vector and store the state of the Observer
* needs synchronization, but notifying observers
* does not (should not). The worst result of any
* potential race-condition here is that:
* 1) a newly-added Observer will miss a
* notification in progress
* 2) a recently unregistered Observer will be
* wrongly notified when it doesn't care
*/
if (!changed)
return;
arrLocal = obs.toArray();
clearChanged();
}
for (int i = arrLocal.length-1; i>=0; i--)
((Observer)arrLocal[i]).update(this, arg);
}
/**
* Clears the observer list so that this object no longer has any observers.
*/
public synchronized void deleteObservers() {
obs.removeAllElements();
}
/**
* Marks this <tt>Observable</tt> object as having been changed; the
* <tt>hasChanged</tt> method will now return <tt>true</tt>.
*/
protected synchronized void setChanged() {
changed = true;
}
/**
* Indicates that this object has no longer changed, or that it has
* already notified all of its observers of its most recent change,
* so that the <tt>hasChanged</tt> method will now return <tt>false</tt>.
* This method is called automatically by the
* <code>notifyObservers</code> methods.
*
* @see java.util.Observable#notifyObservers()
* @see java.util.Observable#notifyObservers(java.lang.Object)
*/
protected synchronized void clearChanged() {
changed = false;
}
/**
* Tests if this object has changed.
*
* @return <code>true</code> if and only if the <code>setChanged</code>
* method has been called more recently than the
* <code>clearChanged</code> method on this object;
* <code>false</code> otherwise.
* @see java.util.Observable#clearChanged()
* @see java.util.Observable#setChanged()
*/
public synchronized boolean hasChanged() {
return changed;
}
/**
* Returns the number of observers of this <tt>Observable</tt> object.
*
* @return the number of observers of this object.
*/
public synchronized int countObservers() {
return obs.size();
}
}
1.2.2 观察者
package java.util;
/**
* A class can implement the <code>Observer</code> interface when it
* wants to be informed of changes in observable objects.
*
* @author Chris Warth
* @see java.util.Observable
* @since JDK1.0
*/
public interface Observer {
/**
* This method is called whenever the observed object is changed. An
* application calls an <tt>Observable</tt> object's
* <code>notifyObservers</code> method to have all the object's
* observers notified of the change.
*
* @param o the observable object.
* @param arg an argument passed to the <code>notifyObservers</code>
* method.
*/
void update(Observable o, Object arg);
}
2. guava
- 单机事件的一个很好的实践
- 并不走jdk的事件规范
- 考虑的比较完善
2.1 EventBus
文章来源：https://uudwc.com/A/yk0Em
- Executor: 异步分发事件 使用的线程池
- SubscriberExceptionHandler: 订阅者的异常处理类
- Dispatcher：根据规则分发
- Subscribe: 订阅者(相当于监听器)
- SubscriberRegsitry: 维护订阅关系
- EventBus: 怎么说呢，事件总线
package com.google.common.eventbus;
// 标记:尚在测试阶段
@Beta
public class EventBus {
private final String identifier;
private final Executor executor;
private final SubscriberExceptionHandler exceptionHandler;
// 注册器持有当前总线的引用
private final SubscriberRegistry subscribers = new SubscriberRegistry(this);
private final Dispatcher dispatcher;
/** Creates a new EventBus named "default". */
public EventBus() {
this("default");
}
public EventBus(String identifier) {
this(
identifier,
// 默认的线程池
// 从命名可知:直接调用的线程池，即单线程的池
MoreExecutors.directExecutor(),
Dispatcher.perThreadDispatchQueue(),
// 异常处理类:日志输出
LoggingHandler.INSTANCE);
}
/**
* Creates a new EventBus with the given {@link SubscriberExceptionHandler}.
*
* @param exceptionHandler Handler for subscriber exceptions.
* @since 16.0
*/
public EventBus(SubscriberExceptionHandler exceptionHandler) {
this(
"default",
MoreExecutors.directExecutor(),
Dispatcher.perThreadDispatchQueue(),
exceptionHandler);
}
EventBus(
String identifier,
Executor executor,
Dispatcher dispatcher,
SubscriberExceptionHandler exceptionHandler) {
this.identifier = checkNotNull(identifier);
this.executor = checkNotNull(executor);
this.dispatcher = checkNotNull(dispatcher);
this.exceptionHandler = checkNotNull(exceptionHandler);
}
// 异常处理，即 捕获异常，调用异常处理类
void handleSubscriberException(Throwable e, SubscriberExceptionContext context) {
checkNotNull(e);
checkNotNull(context);
try {
exceptionHandler.handleException(e, context);
} catch (Throwable e2) {
// if the handler threw an exception... well, just log it
logger.log(
Level.SEVERE,
String.format(Locale.ROOT, "Exception %s thrown while handling exception: %s", e2, e),
e2);
}
}
// 将订阅者注册到register
public void register(Object object) {
// SubscriberRegistry.registry(listener)
subscribers.register(object);
}
public void unregister(Object object) {
subscribers.unregister(object);
}
// 投递事件的地方!
public void post(Object event) {
// 从register中获取匹配事件的、已注册的订阅者
Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);
if (eventSubscribers.hasNext()) {
// step into ...
// 准备分发到对应的订阅者
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));
}
}
// 默认的输出日志的异常处理类
static final class LoggingHandler implements SubscriberExceptionHandler {
static final LoggingHandler INSTANCE = new LoggingHandler();
@Override
public void handleException(Throwable exception, SubscriberExceptionContext context) {
Logger logger = logger(context);
if (logger.isLoggable(Level.SEVERE)) {
logger.log(Level.SEVERE, message(context), exception);
}
}
private static Logger logger(SubscriberExceptionContext context) {
return Logger.getLogger(EventBus.class.getName() + "." + context.getEventBus().identifier());
}
private static String message(SubscriberExceptionContext context) {
Method method = context.getSubscriberMethod();
return "Exception thrown by subscriber method "
+ method.getName()
+ '('
+ method.getParameterTypes()[0].getName()
+ ')'
+ " on subscriber "
+ context.getSubscriber()
+ " when dispatching event: "
+ context.getEvent();
}
}
}
2.1.1 MoreExecutors.directExecutor() 默认的线程池
// com.google.common.util.concurrent.MoreExecutors#directExecutor
public static Executor directExecutor() {
// step into ...
return DirectExecutor.INSTANCE;
}
// 枚举单例
// com.google.common.util.concurrent.DirectExecutor
@GwtCompatible
enum DirectExecutor implements Executor {
INSTANCE;
@Override
public void execute(Runnable command) {
// 直接执行，不绕圈子
command.run();
}
}
2.1.2 Dispatcher.perThreadDispatchQueue() 默认的事件分发器
文章来源地址https://uudwc.com/A/yk0Em
// com.google.common.eventbus.Dispatcher#perThreadDispatchQueue
static Dispatcher perThreadDispatchQueue() {
return new PerThreadQueuedDispatcher();
}
// com.google.common.eventbus.Dispatcher.PerThreadQueuedDispatcher
private static final class PerThreadQueuedDispatcher extends Dispatcher {
// This dispatcher matches the original dispatch behavior of EventBus.
/** Per-thread queue of events to dispatch. */
private final ThreadLocal<Queue<Event>> queue =
new ThreadLocal<Queue<Event>>() {
@Override
protected Queue<Event> initialValue() {
return Queues.newArrayDeque();
}
};
/** Per-thread dispatch state, used to avoid reentrant event dispatching. */
private final ThreadLocal<Boolean> dispatching =
new ThreadLocal<Boolean>() {
@Override
protected Boolean initialValue() {
return false;
}
};
// 分发逻辑在此
@Override
void dispatch(Object event, Iterator<Subscriber> subscribers) {
checkNotNull(event);
checkNotNull(subscribers);
Queue<Event> queueForThread = queue.get();
queueForThread.offer(new Event(event, subscribers));
if (!dispatching.get()) {
dispatching.set(true);
try {
Event nextEvent;
while ((nextEvent = queueForThread.poll()) != null) {
while (nextEvent.subscribers.hasNext()) {
nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
}
}
} finally {
dispatching.remove();
queue.remove();
}
}
}
private static final class Event {
private final Object event;
private final Iterator<Subscriber> subscribers;
private Event(Object event, Iterator<Subscriber> subscribers) {
this.event = event;
this.subscribers = subscribers;
}
}
}
