Hadoop的心脏：中央异步调度器AsyncDispatcher代码和设计解析

以Yarn、HDFS和MapReduce为主要组成的Hadoop，涉及到大量复杂的、交互的事件处理、状态转换，同时，这些事件调度和状态转换又对实时性和效率提出了极高的要求。可以想见，没有一个规整的、通用型良好的调度器，Hadoop代码无论是对读者，还是对开发者，都将变成一场灾难，同时，Hadoop的运行效率也会变得无法忍受。统一的、设计良好的、通用的和共用的调度器，对于Hadoop不同组件的开发者来说是一种解脱，大大降低了Hadoop在事件调度、状态转换的底层出错的可能性，提高了代码稳定性和可读性。这篇文章主要介绍了Hadoop的核心调度器AsyncDispatcher的设计和实现。同Hadoop状态机一样，这个通用调度器设计得十分通用，完美可扩展可重用，我们在自己的项目中完全可以使用Hadoop的调度器实现我们自己的事件调度逻辑。

先抛开代码本身，我们来看一个基于事件的调度器的工作方式：

public AsyncDispatcher() {
    this(new LinkedBlockingQueue<Event>());
  }

  public AsyncDispatcher(BlockingQueue<Event> eventQueue) {
    super("Dispatcher");
    this.eventQueue = eventQueue;
    this.eventDispatchers = new HashMap<Class<? extends Enum>, EventHandler>();
  }
           

在它的构造方法中，有两个核心变量，eventQueue是一个队列，用来存放事件，同时，还构造了一个eventDispatchers，这是一个map对象，用来管理事件和事件处理器之间的关系。AsyncDispatcher同样遵循我一直提到的Yarn的服务化设计，即AsyncDispatcher被抽象为一个service，在完成对象构造以后，会进行服务的初始化和启动：

@Override
  protected void serviceInit(Configuration conf) throws Exception {
    this.exitOnDispatchException =
       conf.getBoolean(Dispatcher.DISPATCHER_EXIT_ON_ERROR_KEY,
          Dispatcher.DEFAULT_DISPATCHER_EXIT_ON_ERROR);
    super.serviceInit(conf);
  }
           

初始化操作并没有做太多的事情，只是调用父类的serviceInit()方法，用来将配置文件设置进来。

@Override
  protected void serviceStart() throws Exception {
    //start all the components
    super.serviceStart();
    eventHandlingThread = new Thread(createThread());
    eventHandlingThread.setName("AsyncDispatcher event handler");
    eventHandlingThread.start();
  }
           

serviceStart()方法的主要任务是创建了一个名字叫做

AsyncDispatcher event handler

的独立线程，我们一起来看这个线程的代码：

Runnable createThread() {
    return new Runnable() {
      @Override
      public void run() {
        while (!stopped && !Thread.currentThread().isInterrupted()) {
          drained = eventQueue.isEmpty();
          // blockNewEvents is only set when dispatcher is draining to stop,
          // adding this check is to avoid the overhead of acquiring the lock
          // and calling notify every time in the normal run of the loop.
          if (blockNewEvents) {
            synchronized (waitForDrained) {
              if (drained) {
                waitForDrained.notify();
              }
            }
          }
          Event event;
          try {
            event = eventQueue.take();
          } catch(InterruptedException ie) {
            if (!stopped) {
              LOG.warn("AsyncDispatcher thread interrupted", ie);
            }
            return;
          }
          if (event != null) {
            dispatch(event);
          }
        }
      }
    };
  }
           

线程的大致功能，是不断循环检测事件队列eventQueue中是否有新的事件到来，如果有，则取出这个事件的处理器，一个EventHandler接口的实现类，调用dispatch()方法，对该事件进行处理。

那么问题来了，这个异步调度器会管理多个事件和事件调度器，那么它什么时候开始知道某个事件发生的时候改用什么调度器对象对这个事件进行处理呢？这就是register方法：

/**
   * 注册事件分派器
   */
  public void register(Class<? extends Enum> eventType,
      EventHandler handler) {
    /* check to see if we have a listener registered */
    EventHandler<Event> registeredHandler = (EventHandler<Event>)
    eventDispatchers.get(eventType);
    LOG.info("Registering " + eventType + " for " + handler.getClass());
    if (registeredHandler == null) {
      eventDispatchers.put(eventType, handler);
    } else if (!(registeredHandler instanceof MultiListenerHandler)){
      /* for multiple listeners of an event add the multiple listener handler */
      MultiListenerHandler multiHandler = new MultiListenerHandler();
      multiHandler.addHandler(registeredHandler);
      multiHandler.addHandler(handler);
      eventDispatchers.put(eventType, multiHandler);
    } else {
      /* already a multilistener, just add to it */
      MultiListenerHandler multiHandler
      = (MultiListenerHandler) registeredHandler;
      multiHandler.addHandler(handler);
    }
  }
           

还有一个问题，不同的事件，是怎么放入eventQueue中的呢？这是通过AsyncDispatcher.GenericEventHandler.handle()方法进行的：

class GenericEventHandler implements EventHandler<Event> {
    public void handle(Event event) {
      if (blockNewEvents) { //如果该标记为置位，说明可能服务正在进行stop操作，无法处理新的请求
        return;
      }
      drained = false;
      /* all this method does is enqueue all the events onto the queue */
      int qSize = eventQueue.size();
      if (qSize !=  && qSize %  == 
          && lastEventQueueSizeLogged != qSize) {
        lastEventQueueSizeLogged = qSize;
        LOG.info("Size of event-queue is " + qSize);
      }
      int remCapacity = eventQueue.remainingCapacity();
      if (remCapacity < ) {
        LOG.warn("Very low remaining capacity in the event-queue: "
            + remCapacity);
      }
      try {
        eventQueue.put(event);
      } catch (InterruptedException e) {
        if (!stopped) {
          LOG.warn("AsyncDispatcher thread interrupted", e);
        }
        // Need to reset drained flag to true if event queue is empty,
        // otherwise dispatcher will hang on stop.
        drained = eventQueue.isEmpty();
        throw new YarnRuntimeException(e);
      }
    };
  }
           

AsyncDispatcher

内部维护了一个

GenericEventHandler handlerInstance

变量，通过z这个

handlerInstance.handle()

来将新的事件放入到

eventQueue

中。这个方法名字虽然叫做

handle()

，但是由于我们的AysncDispatcher是异步调度器，因此并没有立刻进行处理，而是放入队列，由上面提到的

eventHandlingThread

线程进行处理。

先看AsyncDispatcher.serviceStop()方法：

protected void serviceStop() throws Exception {
    if (drainEventsOnStop) {
      blockNewEvents = true; //首先阻止新任务的分派，试图优雅停掉当前线程的工作
      LOG.info("AsyncDispatcher is draining to stop, igonring any new events.");
      long endTime = System.currentTimeMillis() + getConfig()
          .getLong(YarnConfiguration.DISPATCHER_DRAIN_EVENTS_TIMEOUT,
              YarnConfiguration.DEFAULT_DISPATCHER_DRAIN_EVENTS_TIMEOUT);

      synchronized (waitForDrained) {
        while (!drained && eventHandlingThread != null
            && eventHandlingThread.isAlive()
            && System.currentTimeMillis() < endTime) {
          waitForDrained.wait();
          LOG.info("Waiting for AsyncDispatcher to drain. Thread state is :" +
              eventHandlingThread.getState());
        }
      }
    }
    stopped = true;
    if (eventHandlingThread != null) {
      eventHandlingThread.interrupt();//防止线程正在处理一个耗时任务导致线程依然没有退出
      try {
        eventHandlingThread.join();//等待eventHandlingThread执行完毕
      } catch (InterruptedException ie) {
        LOG.warn("Interrupted Exception while stopping", ie);
      }
    }
           

serviceStop()调用一开始，就将blockNewEvents=true置位，这意味着将不再接受新的event提交，然后的主要工作，就是需要等待所有的serviceStop()调用前提交的事件都被执行完毕再真正停止服务，否则会造成任务丢失。

waitForDrained.wait(1000);

代表着服务关闭的时候会每1s检查我们的eventQueue中的所有事件是否处理完毕，如果没有处理完毕，则继续等待。虽然1s已经非常短暂，但是对于Yarn这样的高并发系统，也必须进行优化。

为了能够在eventQueue清空的时候及时通知serviceStop()所在线程，我们看这段代码：

if (blockNewEvents) {
            synchronized (waitForDrained) {
              if (drained) {
                waitForDrained.notify();
              }
            }
}
           

如果

blockNewEvents=true

，代表该调度器目前正在进行关闭操作，因此，事件处理线程有必要在发现事件队列已经清空的情况下(drained=true)，通过waitForDrained锁及时通知到服务关闭线程(waitForDrained.notify())。服务关闭线程收到该通知就能立刻从waitForDrained.wait()中直接唤醒，从而及时完成关闭操作，而不再进行不必要的wait操作。

注意，wait操作与sleep()不同，wait()操作虽然必须在synchronized代码块中，但是等待过程中会放弃锁。因此，serviceStop()正在执行wait()，队列处理线程

AsyncDispatcher event handler

依然可以进入synchronized(waitForDrained)检查eventQueue是否已经完全清空。

由此可见这段的其实很关键，目的是为了在某些情况下我们需要关闭AsyncDispatcher的时候，能够非常优雅的、消息无丢失地同时迅速的关闭服务。

AsyncDispatcher event handler

线程会不断从eventQueue中取出事件，然后交付给分派方法dispatch()对这个事件进行派发：

Event event;
          try {
            event = eventQueue.take();
          } catch(InterruptedException ie) {
            if (!stopped) {
              LOG.warn("AsyncDispatcher thread interrupted", ie);
            }
            return;
          }
          if (event != null) {
            dispatch(event);
          }
           

我们来看

dispatch()

方法：

protected void dispatch(Event event) {
    //all events go thru this loop
    if (LOG.isDebugEnabled()) {
      LOG.debug("Dispatching the event " + event.getClass().getName() + "."
          + event.toString());
    }

    Class<? extends Enum> type = event.getType().getDeclaringClass();

    try{
      EventHandler handler = eventDispatchers.get(type);
      if(handler != null) {
        handler.handle(event);
      } else {
        throw new Exception("No handler for registered for " + type);
      }
    } catch (Throwable t) {
      //TODO Maybe log the state of the queue
      LOG.fatal("Error in dispatcher thread", t);
      // If serviceStop is called, we should exit this thread gracefully.
      if (exitOnDispatchException
          && (ShutdownHookManager.get().isShutdownInProgress()) == false
          && stopped == false) {
        Thread shutDownThread = new Thread(createShutDownThread());
        shutDownThread.setName("AsyncDispatcher ShutDown handler");
        shutDownThread.start();
      }
    }
  }
           

代码比较简单，就是根据事件类型

event.getType().getDeclaringClass();

,从

eventDispatchers

中取出对应的EventHandler，然后调用

handle()

方法对事件进行处理。从

eventDispatchers

的声明

protected final Map<Class<? extends Enum>, EventHandler> eventDispatchers

中可以看到，

eventDispatchers

的key是

Class<? extends Enum>

，即是一个

Enum

对象的子类的Class类型。

我们可以从

Hadoop-Yarn

的一个使用场景来看看Hadoop是如何使用

Asynchronized

进行事件调度的。我们以

Yarn ResourceManager

为例，来看看它是如何使用中央异步调度器进行事件调度的。

同样，

ResourceManager

也被Yarn抽象为服务，在

ResourceManager.serviceInit()

方法中，创建了调度器对象：

private Dispatcher setupDispatcher() {
    Dispatcher dispatcher = createDispatcher();
    dispatcher.register(RMFatalEventType.class,
        new ResourceManager.RMFatalEventDispatcher());
    return dispatcher;
  }
           

protected Dispatcher createDispatcher() {
    return new AsyncDispatcher();
  }
           

在

ResourceManager

服务的子服务

RMActiveServices

的初始化方法

serviceInit()

中，在这个调度器对象上注册了各种事件和handler：

rmDispatcher.register(RMAppEventType.class,
          new ApplicationEventDispatcher(rmContext));

      // Register event handler for RmAppAttemptEvents
      //将RmAppAttemptEvent交给RMAppAttempt去维护，其实现类是RMAppAttemptImpl
      rmDispatcher.register(RMAppAttemptEventType.class,
          new ApplicationAttemptEventDispatcher(rmContext));

      // Register event handler for RmNodes
      //将RmNodesEvent交给RMNode去维护，其实现类是RMNodeImpl
      rmDispatcher.register(
          RMNodeEventType.class, new NodeEventDispatcher(rmContext));
           

我们以

RMAppAttemptEventType

类型的事件进行分析。注意，一个类型是指相同事件的集合，

RMAppAttemptEventType

这个类型的事件代表了与

ResouceManager App Attempt

(即ApplicationMaster本身的attempt进程)相关的事件集合，如

RMAppAttemptEventType.START

，

RMAppAttemptEventType.KILL

，

RMAppAttemptEventType.LAUNCHED

等等，都是

enum

类型。

在ApplicationMasterService.allocate()方法中（关于ApplicaitonMaster与ApplicationMasterService之间的通信，大家可以看我的博客[YARN ApplicationMaster与ResourceManager之间基于applicationmaster_protocol.proto协议的allocate()接口源码解析]），需要处理

RMAppAttemptEventType

类型的事件：

//this.rmContext.getDispatcher().getEventHandler()返回内部维护的GenericEventHandler handlerInstance
this.rmContext.getDispatcher().getEventHandler().handle(
          new RMAppAttemptStatusupdateEvent(appAttemptId, request
              .getProgress()));
           

显然，当前发生的事件是

RMAppAttemptStatusupdateEvent

，

GenericEventHandler handlerInstance

在

handle()

方法中将该事件放入

eventQueue

，

eventHandlingThread

从

RMAppAttemptStatusupdateEvent

中取出

eventType

，我们跟踪一下

RMAppAttemptStatusupdateEvent

的代码可以看出，

eventType

正是

RMAppAttemptEventType

。好吧，刚才说过，在

ResourceManager

服务的子服务

RMActiveServices

的初始化方法

serviceInit()

中已经为这个

RMAppAttemptEventType

注册了handler，因此取出这个handler来处理这个事件即可。

以上就是异步调度器

AsyncDispatcher

的整个工作流程，良好的通用型和异步设计，使得Hadoop几乎所有的事件调度都使用它来进行。这种基于事件的设计思想对于一个高并发的系统来说非常实用。