《Apache Shiro 源码解析》- 9.事件总线

import org.apache.shiro.event.Event;

/**
 * 自定义事件类
 */
public class CustomEvent extends Event {
    private final String message;

    public CustomEvent(String message) {
        super(new Object()); // source 是事件的触发对象
        this.message = message;
    }

    public String getMessage() {
        return message;
    }
}

/**
 * 实现事件监听器
 */
public class CustomEventListener implements SingleArgumentMethodEventListener<CustomEvent> {
    @Override
    public void onEvent(CustomEvent event) {
        System.out.println("Received event: " + event.getMessage());
    }

    @Override
    public boolean accepts(Event event) {
        return event instanceof CustomEvent;
    }
}

/**
 * 使用 EventBus 发布事件
 */
public class EventBusExample {
    public static void main(String[] args) {
        EventBus eventBus = new EventBus();

        // 注册事件监听器
        eventBus.register(new CustomEventListener());

        // 发布事件
        eventBus.publish(new CustomEvent("Hello, Shiro EventBus!"));
    }
}

import org.apache.shiro.event.Subscribe;

public class CustomEventListener  {
    @Subscribe
    public void onCustomEvent(CustomEvent event) {
        System.out.println("Received event: " + event.getMessage());
    }
}

public interface EventBus {
   /**
    * 该方法用于发布事件，事件会被广播给所有订阅者。
    */
    void publish(Object event);
    /**
     * 该方法用于注册一个订阅者对象。
     */
    void register(Object subscriber);
    /**
     * 该方法用于取消一个订阅者对象的注册，从而停止接收事件。如果该对象之前未被注册，则调用此方法无效果。
     */
    void unregister(Object subscriber);
}

public class DefaultEventBus implements EventBus {
    //...
    private EventListenerResolver eventListenerResolver;

    //这是一个“注册表”，所有订阅者都会被保存到这个数据结构中。
    private final Map<Object, Subscription> registry;

    /**
     * 构造方法
     */
    public DefaultEventBus() {
        // 创建“注册表”实例
        this.registry = new LinkedHashMap<Object, Subscription>();

        //...
        this.eventListenerResolver = new AnnotationEventListenerResolver();
    }

    /**
     * 发布事件
     */
    public void publish(Object event) {
        if (event == null) {
            log.info("Received null event for publishing.  Ignoring and returning.");
            return;
        }

        //...
        try {
            //注意这里：当事件发生的时候，会遍历所有订阅者，调用上面的 onEvent 方法，并且把事件对象传递进去（广播）。
            for (Subscription subscription : this.registry.values()) {
                subscription.onEvent(event);
            }
        } finally {
            registryReadLock.unlock();
        }
    }

    /**
     * 把事件订阅者注册到“注册表”中
     */
    public void register(Object instance) {
        if (instance == null) {
            log.info("Received null instance for event listener registration.  Ignoring registration request.");
            return;
        }

        //注意这里的细节，先尝试取消注册，防止同一个订阅者被重复注册多次。
        unregister(instance);

        List<EventListener> listeners = getEventListenerResolver().getEventListeners(instance);

        if (listeners == null || listeners.isEmpty()) {
            log.warn("Unable to resolve event listeners for subscriber instance [{}]. Ignoring registration request.",
                    instance);
            return;
        }

        Subscription subscription = new Subscription(listeners);

        //...
        try {
          //把订阅者实例 put 到 Map 中
          this.registry.put(instance, subscription);
        } finally {
          //...
        }
    }

    /**
     * 取消订阅
     */
    public void unregister(Object instance) {
        if (instance == null) {
            return;
        }

        //...

        try {
            //从“注册表”中删除，这样就收不到事件了。
            this.registry.remove(instance);
        } finally {
            //...
        }
    }
}

private class Subscription {
    //维护一个事件监听器列表
    private final List<EventListener> listeners;

    /**
     * 构造方法
     */
    public Subscription(List<EventListener> listeners) {
        List<EventListener> toSort = new ArrayList<EventListener>(listeners);
        //排序
        Collections.sort(toSort, EVENT_LISTENER_COMPARATOR);
        this.listeners = toSort;
    }

    /**
     * 当事件发生的时候，会调用此方法
     */
    public void onEvent(Object event) {

        //保存已经发布过事件的对象，防止在同一个对象上重复派发事件。
        Set<Object> delivered = new HashSet<Object>();

        //遍历所有事件监听器，开始给它们派发事件。
        for (EventListener listener : this.listeners) {
            Object target = listener;
            if (listener instanceof SingleArgumentMethodEventListener) {
                SingleArgumentMethodEventListener singleArgListener = (SingleArgumentMethodEventListener) listener;
                target = singleArgListener.getTarget();
            }

            //先判断监听器是否能接受当前的事件对象（这里的本质是单播，避免把不相关的事件派发给监听器）
            if (listener.accepts(event) && !delivered.contains(target)) {
                try {
                    //关键：调用事件监听器上的 onEvent 方法。
                    listener.onEvent(event);
                } catch (Throwable t) {
                    log.warn(EVENT_LISTENER_ERROR_MSG, t);
                }
                delivered.add(target);
            }
        }
    }
}

public void onEvent(Object event) {
    Method method = getMethod();
    try {
        //这里借助于 Java 的反射机制，进行方法调用
        method.invoke(getTarget(), event);
    } catch (Exception e) {
        throw new IllegalStateException("Unable to invoke event handler method [" + method + "].", e);
    }
}

List<EventListener> listeners = getEventListenerResolver().getEventListeners(instance);

public List<EventListener> getEventListeners(Object instance) {
    if (instance == null) {
        return Collections.emptyList();
    }

    //利用 Java 的反射机制，获取所有添加了事件注解的方法。
    List<Method> methods = ClassUtils.getAnnotatedMethods(instance.getClass(), getAnnotationClass());
    if (methods == null || methods.isEmpty()) {
        return Collections.emptyList();
    }

    List<EventListener> listeners = new ArrayList<EventListener>(methods.size());

    for (Method m : methods) {
        listeners.add(new SingleArgumentMethodEventListener(instance, m));
    }

    return listeners;
}

@Retention(value = RetentionPolicy.RUNTIME) //运行时注解
@Target(value = ElementType.METHOD)         //只能装饰方法
@Documented
public @interface Subscribe {
}

public class DefaultEventBus implements EventBus {
    //...

    private final Map<Object, Subscription> registry;
    //读取锁
    private final Lock registryReadLock;
    //写入锁
    private final Lock registryWriteLock;

    public DefaultEventBus() {
        //注意， LinkedHashMap 不是一个线程安全的类型
        //也就是说，如果多个线程同时操作 LinkedHashMap ，不能保证数据的一致性，也不能保证不出现数据竞争的情况
        this.registry = new LinkedHashMap<Object, Subscription>();

        //读写锁
        ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
        this.registryReadLock = rwl.readLock();
        this.registryWriteLock = rwl.writeLock();
    }

    public void publish(Object event) {
        //...

        //注意这里，先加读取锁
        registryReadLock.lock();

        try {
            //...
        } finally {
            //释放读取锁
            registryReadLock.unlock();
        }
    }

    public void register(Object instance) {
        //...

        //注意这里，先加上写入锁
        this.registryWriteLock.lock();
        try {
            //...
        } finally {
            //释放写入锁
            this.registryWriteLock.unlock();
        }
    }

    public void unregister(Object instance) {
        //...

        //注意这里，先加写入锁
        this.registryWriteLock.lock();
        try {
            //...
        } finally {
            //释放写入锁
            this.registryWriteLock.unlock();
        }
    }

    //...
}

public void execute() {

    for( Statement statement : statements) {

        statement.execute();

        BeanConfiguration bd = statement.getBeanConfiguration();

        if (bd.isExecuted()) {
            if (bd.getBeanName().equals(EVENT_BUS_NAME)) {
                EventBus eventBus = (EventBus)bd.getBean();
                enableEvents(eventBus);
            }

            if (!bd.isGlobalConfig()) {
                BeanEvent event = new ConfiguredBeanEvent(bd.getBeanName(), bd.getBean(),
                        Collections.unmodifiableMap(objects));

                //注意这里，在事件总线上派发了一个事件
                eventBus.publish(event);
            }

            LifecycleUtils.init(bd.getBean());

            if (!bd.isGlobalConfig()) {
                BeanEvent event = new InitializedBeanEvent(bd.getBeanName(), bd.getBean(),
                        Collections.unmodifiableMap(objects));

                //注意这里，在事件总线上派发了一个事件
                eventBus.publish(event);
            }
        }
    }
}

public class EventObject implements java.io.Serializable {
    //...

    protected transient Object source;

    public EventObject(Object source) {
        if (source == null)
            throw new IllegalArgumentException("null source");

        //注意这里，EventObject 的实现非常简单，只持有了一个事件源对象的实例。
        this.source = source;
    }

    //...
}

public abstract class BeanEvent extends Event {

    private String beanName;
    private Object bean;
    private final Map<String, Object> beanContext;

    public BeanEvent(final String beanName, final Object bean, final Map<String, Object> beanContext) {
        super(bean);
        this.beanName = beanName;
        this.bean = bean;
        this.beanContext = beanContext;
    }

    //...
}

public class InitializedBeanEvent extends BeanEvent {
    public InitializedBeanEvent(String beanName, Object bean, Map<String, Object> beanContext) {
        super(beanName, bean, beanContext);
    }
}