共计 4297 个字符，预计需要花费 11 分钟才能阅读完成。
前言
今天，我们将介绍一个比较新的设计模式（也就是没有在 GoF 那本书中出现过的一种设计模式），这个设计模式就是 Event Bus 设计模式。
起源
假设一个大型应用中，有大量的组件彼此间存在交互。而你希望能够在组件通信的同时能够满足低耦合和关注点分离原则。Event Bus 设计模式是一个很好的解决方案。
Event Bus 的概念和网络中的总线拓扑概念类似。即存在某种管道，而所有的电脑都连接在这条管道之上。其中的任何一台电脑发送的消息都将分发给总线上所有其它的主机。然后，每台主机决定是否接收还是抛弃掉这条消息。
在组件的层面上也是类似的：主机对应着应用的组件，消息对应于事件（event）或者数据。而管道是 Event Bus 对象。
实现
并没有一种绝对正确的实现 EventBus 的方式。在这里我会简单的介绍两种方法。
第一种方法
这是比较经典的一种方法，它主要取决于定义 EventBus 接口（从而强制实现一个特定的协议），按需对其实现，然后定义一个 Subscriber（另一份协议 ) 来进行 Event 的处理。
/**
* interface describing a generic event, and it’s associated meta data, it’s this what’s going to
* get sent in the bus to be dispatched to intrested Subscribers
*
* @author chermehdi
*/
public interface Event<T> {
/**
* @returns the stored data associated with the event
*/
T getData();
}
import java.util.Set;
/**
* Description of a generic subscriber
*
* @author chermehdi
*/
public interface Subscribable {
/**
* Consume the events dispatched by the bus, events passed as parameter are can only be of type
* declared by the supports() Set
*/
void handle(Event<?> event);
/**
* describes the set of classes the subscribable object intends to handle
*/
Set<Class<?>> supports();
}
import java.util.List;
/**
* Description of the contract of a generic EventBus implementation, the library contains two main
* version, Sync and Async event bus implementations, if you want to provide your own implementation
* and stay compliant with the components of the library just implement this contract
*
* @author chermehdi
*/
public interface EventBus {
/**
* registers a new subscribable to this EventBus instance
*/
void register(Subscribable subscribable);
/**
* send the given event in this EventBus implementation to be consumed by interested subscribers
*/
void dispatch(Event<?> event);
/**
* get the list of all the subscribers associated with this EventBus instance
*/
List<Subscribable> getSubscribers();
}
Subscribable 接口定义了一个方法来处理一个特定类型的消息，并且通过定义 supports 方法决定支持哪种类型。
EventBus 的实现持有所有 Subscribable 对象，并且每当一个新事件触发 dispatch 方法时，通知所有的 Subscribable 对象。
这种方案的有点事可以在编译时检查传递过来的 Subscribable 对象，而且更加符合面向对象的思想，因为无需使用反射。同时，可以看到，这种方案更容易实现。缺点是接口的强制性 – 你总是需要一个新的类来处理一种类型的 Event，在项目初期这个问题可能不明显，但是随着项目发展，你会发现，新建一个类只是为了处理简单的逻辑比如日志或是数据分析会显得很冗余。
第二种方法
这个方法来源于 Guava 的实现。EventBus 看上去更简单更好用，对于每个时间的 consumer，你只需要通过对一个方法加上 @Subscribe 注解，并且在注解的参数中传入你希望处理的对象类型（单个对象 / 参数）。然后你通过调用 eventBus.register(objectContainingTheMethod) 来注册事件的消费者。要产生一个新的时间，你只需要调用 eventBus.post(someObject)，然后所有相关的消费者都将会被通知。
如果对应一个特定的对象没有对应的消费者怎么办？在 guava 的实现中，它们被称为 DeadEvents，在我的实现中，post 调用会被忽略。
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
/**
* Simple implementation demonstrating how a guava EventBus works generally, without all the noise
* of special cases handling, and special guava collections
*
* @author chermehdi
*/
public class EventBus {
private Map<Class<?>, List<Invocation>> invocations;
private String name;
public EventBus(String name) {
this.name = name;
invocations = new ConcurrentHashMap<>();
}
public void post(Object object) {
Class<?> clazz = object.getClass();
if (invocations.containsKey(clazz)) {
invocations.get(clazz).forEach(invocation -> invocation.invoke(object));
}
}
public void register(Object object) {
Class<?> currentClass = object.getClass();
// we try to navigate the object tree back to object ot see if
// there is any annotated @Subscribe classes
while (currentClass != null) {
List<Method> subscribeMethods = findSubscriptionMethods(currentClass);
for (Method method : subscribeMethods) {
// we know for sure that it has only one parameter
Class<?> type = method.getParameterTypes()[0];
if (invocations.containsKey(type)) {
invocations.get(type).add(new Invocation(method, object));
} else {
List<Invocation> temp = new Vector<>();
temp.add(new Invocation(method, object));
invocations.put(type, temp);
}
}
currentClass = currentClass.getSuperclass();
}
}
private List<Method> findSubscriptionMethods(Class<?> type) {
List<Method> subscribeMethods = Arrays.stream(type.getDeclaredMethods())
.filter(method -> method.isAnnotationPresent(Subscribe.class))
.collect(Collectors.toList());
checkSubscriberMethods(subscribeMethods);
return subscribeMethods;
}
private void checkSubscriberMethods(List<Method> subscribeMethods) {
boolean hasMoreThanOneParameter = subscribeMethods.stream()
.anyMatch(method -> method.getParameterCount() != 1);
if (hasMoreThanOneParameter) {
throw new IllegalArgumentException(
“Method annotated with @Susbscribe has more than one parameter”);
}
}
public Map<Class<?>, List<Invocation>> getInvocations() {
return invocations;
}
public String getName() {
return name;
}
}
可以看到这种方案所需要的额外工作比较少。你只需要定义方法的名称而不是为各个处理器命名。而且你可以将所有的消费者定义在一个类中。你只需要为每个方法传递不同的事件类型即可。