Azkaban使用的两个设计模式

2、观察者模式

这是一个比较常用的设计模式了，以前在学习linux网络编程的时候就经常接触到这种模式，那时我们的网络编程模型使用的是epoll，并且在同一个连接的所有请求都是异步的，我们为每一个连接注册epoll的读写监听事件，然后由epoll在连接就绪的时候调用，这时候epoll就是一个观察者，它能够观察所有就绪的时间，然后取出保存在epoll中的对象指针调用处理函数来处理。所以在这种模式中涉及到两类对象：观察者和处理者，观察者负责监听所有可能发生的时间，在事件就绪之后通知处理者（一般通过调用处理者的接口），在处理者需要注册到观察者，当然可以在观察这种保存每个处理者感兴趣的事件再根据发生的事件类型判断是否通知，也可以将全部的事件都通知处理者，由处理者决定对那些事件进行处理，epoll使用的是前者，azkaban使用的是后者。

在azkaban中定义了EventListener 接口，处理者需要实现该接口，根据参数的event对象执行处理程序：

public interface EventListener {
  public void handleEvent(Event event);
}

public class EventHandler {
  private HashSet<EventListener> listeners = new HashSet<EventListener>();

  public EventHandler() {
  }

  //向观察者注册
  public void addListener(EventListener listener) {
    listeners.add(listener);
  }

  //事件触发，通知所有注册的处理者
  public void fireEventListeners(Event event) {
    ArrayList<EventListener> listeners =
        new ArrayList<EventListener>(this.listeners);
    for (EventListener listener : listeners) {
      listener.handleEvent(event);
    }
  }

  //取消注册
  public void removeListener(EventListener listener) {
    listeners.remove(listener);
  }
}

  public enum Type {
    FLOW_STARTED,                   //一个流启动执行
    FLOW_FINISHED,                   //一个流执行完成
    JOB_STARTED,                      //一个任务开始执行
    JOB_FINISHED,                      //一个任务执行完成
    JOB_STATUS_CHANGED,        //任务状态变化
    EXTERNAL_FLOW_UPDATED,    //内部流更新
    EXTERNAL_JOB_UPDATED        //内部任务更新
  }

 public class LocalFlowWatcherListener implements EventListener {
    @Override
    public void handleEvent(Event event) {
      if (event.getType() == Type.JOB_FINISHED) {
        if (event.getRunner() instanceof FlowRunner) {
          // 这里是在job没有执行（被取消或者跳过）的情况下由FlowRunner通知的
          Object data = event.getData();
          if (data instanceof ExecutableNode) {
            ExecutableNode node = (ExecutableNode) data;
            handleJobStatusChange(node.getNestedId(), node.getStatus());
          }
        } else if (event.getRunner() instanceof JobRunner) {
          // 这里是在一个job执行完成由jobRunner通知flowRunner，在由FlowRunner通知的。
          JobRunner runner = (JobRunner) event.getRunner();
          ExecutableNode node = runner.getNode();
          System.out.println(node + " looks like " + node.getStatus());
          handleJobStatusChange(node.getNestedId(), node.getStatus());
        }
      } else if (event.getType() == Type.FLOW_FINISHED) {
        stopWatcher();
      }
    }
  }