【Java】事件驱动模型和观察者模式

你有一件事情，作这件事情的过程包含了许多职责单一的子过程。这样的状况及其常见。当这些子过程有以下特色时，咱们应该考虑设计一种合适的框架，让框架来完成一些业务无关的事情，从而使得各个子过程的开发能够专一于本身的业务。

•  这些子过程有必定的执行次序；
•  这些子过程之间须要较灵活的跳转；
•  这些子过程也许须要围绕同一个上下文作操做；

此时能够考虑使用事件驱动的方式来组织这些子过程，此时这些子过程能够被称之为事件处理器（或监听器），而将事件处理器组织起来的管理者，叫作事件中心。最显而易见的实现方式，是观察者模式，或者监听者模式。做为一个例子，考虑一个消息转发系统，它从上游接收消息，而后转发给正确的下游用户。整个过程能够拆分为消息解析、消息存储、消息发送等步骤。

 

事件Event

首先定义事件Event。事件将做为一个基本元素，在处理器和事件中心之间创建其连线。这里为了可以统一处理异常。以及针对异常打出日志，除了业务相关的事件，还增长了异常事件和日志事件。固然相应的也应该新增与之对应的事件处理器。

 1 package me.test.eventcenter;
 2 
 3 /**
 4  * Created by chng on 2015/12/18.
 5  */
 6 public class EventName {
 7 
 8     private final String name;
 9     public EventName(String name) {
10         this.name = name;
11     }
12 
13     public static EventName msg_received = new EventName("msg_received");
14     public static EventName msg_resolved = new EventName("msg_resolved");
15     public static EventName msg_stored = new EventName("msg_stored");
16     public static EventName msg_pushed = new EventName("msg_pushed");
17     public static EventName exception_occured = new EventName("exception_occured");
18     public static EventName end_and_log = new EventName("end_and_log");
19 
20     public String getName() {
21         return name;
22     }
23 }

 

事件处理器 EventHandler

随后，定义一个简单的事件处理器的抽象类，其中包含一个单例的事件中心，每一个处理器经过持有这个事件中心来执行注册本身（即订阅一个事件）和呼起下一个事件的操做。

package me.test.eventcenter.handler;

import me.test.eventcenter.EventCenter;
import org.springframework.beans.factory.InitializingBean;

import javax.annotation.Resource;

/**
 * Created by chng on 2015/12/18.
 */
public abstract class EventHandler implements InitializingBean {
    @Resource
    EventCenter eventCenter;

    public abstract void handle(Object ... param);
}

 

事件中心 EventCenter

有了事件和事件处理器，接下来定义一个事件中心，将两者粘起来。

package me.test.eventcenter;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import me.test.eventcenter.handler.EventHandler;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

/**
 * Created by chng on 2015/12/18.
 */
@Component
public class EventCenter {

    private Map<EventName, List<EventHandler>> regTable = Maps.newHashMap();

    /**
     * 向事件中心广播一个时间，驱使事件中心执行该事件的处理器
     * @param eventName
     * @param param
     */
    public void fire(EventName eventName, Object ... param) {
        System.out.println(eventName.getName());
        List<EventHandler> handlerList = regTable.get(eventName);
        if(CollectionUtils.isEmpty(handlerList)) {
            // log
            return;
        }
        for(EventHandler handler: handlerList) {
            try {
                handler.handle(param);
            } catch (Exception e) {
                fire(EventName.exception_occured, e);
            }
        }
    }

    /**
     * 将本身注册为事件中心的某个事件的处理器
     * @param eventName
     * @param handler
     */
    public void register(EventName eventName, EventHandler handler) {

        List<EventHandler> handlerList = regTable.get(eventName);
        if(null == handlerList) {
            handlerList = Lists.newLinkedList();
        }

        handlerList.add(handler);
        regTable.put(eventName, handlerList);
    }
}

在事件中心中，事件和处理器之间的关系表示为一个HashMap，每一个事件能够被多个处理器监听，而一个处理器只能监听一个事件（这样的关系并不是是固定的，也可在运行时动态地改变）。当呼起一个事件时，事件中心找到该事件的监听者，逐个调用他们的处理方法。将各子模块的执行集中在这里管理，还有两个额外的好处：

1 若是发生异常，则呼起异常处理器。这样，一旦业务模块发生了不起不终止整个过程的时候，不须要本身写try/catch子句，而只须要将异常往上抛，直到抛给框架层，由它来作这些统一的事情。然而这并不意味着各业务模块不折不扣地摆脱了难看的try/catch/finally，运行时发生的异常被catch后，并不是均可以直接END，何去何从仍然视状况而定，直接将异常吞掉也何尝不可能。

2 打日志的活儿交给EventCenter就行了，没人比它更清楚当前执行到了哪一步。而各子模块里面，能够省去许多散布在各处的日志语句。对于散弹式日志的问题，解决方法不止一种，AOP也是个不错的选择。

 

测试

为了让整个过程跑起来，咱们只须要发起一个初始的事件，将全部的事件处理器都依次驱动起来：

/**
 * Created by OurEDA on 2015/12/18.
 */
public class TestEventCenter extends BaseTest {

    @Resource
    EventCenter eventCenter;

    @Test
    public void test() {
        RawMessage rawMessage = new RawMessage("NotifyType: amq");
        rawMessage.setType(RawMessage.MessageType.amq);
        eventCenter.fire(EventName.msg_received, notify);
    }
}

以测试经过为目标，咱们开始定义一系列的EventHandler，并将这些Handler注册到合适的事件上。例如一个消息解析的Handler，对msg_receive事件感兴趣，解析完成后将发起msg_store事件，那么: 

package me.test.eventcenter.handler;

import me.test.eventcenter.*;
import me.test.messagedo.Message;
import me.test.messagedo.RawMessage;
import me.test.resolvers.MsgResolverList;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * Created by chng on 2015/12/18.
 */
@Component
public class MsgResolveHandler extends EventHandler implements InitializingBean {

    @Resource
    private MsgResolverList resolverList;

    @Override
    public void handle(Object... param) {

        /**
         * Resolver
         */
        RawMessage rm = (RawMessage) param[0];
        Message message = resolverList.resolve(rm);
        eventCenter.fire(EventName.msg_resolved, message);
    }

    public void afterPropertiesSet() throws Exception {
        eventCenter.register(EventName.msg_received, this);
    }
}

能够看到，对象在初始阶段把本身(this)注册到了事件中内心。handler方法则只关心如何解析消息，不须要关系别的事情。针对不一样类型的消息，解析器能够写成Map的形式，一种类型对应一个解析器；若是消息的分类比较复杂，还能够写成职责链的形式固然这都可有可无，咱们须要知道的是，这个模块只解析消息，与其余子模块之间是彻底解耦的。

 

例如，一种可能的解析器组合体是这样的：

MsgResolver.java (interface)

package me.test.resolvers;

import me.test.messagedo.Message;
import me.test.messagedo.RawMessage;

/**
 * Created by OurEDA on 2015/12/18.
 */
public interface MsgResolver {

    public boolean canResolve(RawMessage rm);

    public Message resolve(RawMessage rm);

}

MsgResolverList.java

package me.test.resolvers;

import me.test.messagedo.Message;
import me.test.messagedo.RawMessage;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Created by chng on 2015/12/18.
 */
@Component
public class MsgResolverList implements MsgResolver{

    //职责链
    private List<MsgResolver> resolvers;
    public List<MsgResolver> getResolvers() {
        return resolvers;
    }
    public void setResolvers(List<MsgResolver> resolvers) {
        this.resolvers = resolvers;
    }

    public boolean canResolve(RawMessage rawMessage) {
        return true;
    }

    public Message resolve(RawMessage rawMessage) {
        for(MsgResolver resolver: resolvers) {
            if(resolver.canResolve(rawMessage)) {
                System.out.println("NotifyType: "+rawMessage.type);
                return resolver.resolve(rawMessage);
            }
        }
        return null;
    }
}

 没必要额外打日志，用例的输出是这样的：

哪一步出了问题，出了什么问题，统统一目了然。

 

其余：

1 上下文 Context

各个处理器都围绕一个上下文作处理，此例为了体现通用性，上下文直接用Object表示。在实际的场景下，则须要一个统一的结构体。不一样的Handler将对该统一上下文的不一样内容感兴趣。

 

2 线程封闭 ThreadLocal

当有多个线程都在事件中心中进行周转时，还须要考虑线程安全问题，保证线程的调度不会对事件处理器的呼起次序形成干扰。所以整个事件中心和上下文，都须要作隔离。

 

3 反思

上面这种写法有两个明确的缺点：事件的注册操做写死在每一个处理器的初始化代码中，一来缺少灵活性，二来对于各Handler是如何组织起来的，没有一个统一而清晰的bigmap。