Hadoop YARN架构设计要点

public class NMLivelinessMonitor extends AbstractLivelinessMonitor<NodeId>

1    @Override
2    protected void expire(NodeId id) {
3      dispatcher.handle(
4          new RMNodeEvent(id, RMNodeEventType.EXPIRE));
5    }

01    private class PingChecker implements Runnable {
02     
03      @Override
04      public void run() {
05        while (!stopped && !Thread.currentThread().isInterrupted()) {
06          synchronized (AbstractLivelinessMonitor.this) {
07            Iterator<Map.Entry<O, Long>> iterator =
08              running.entrySet().iterator();
09     
10            //avoid calculating current time everytime in loop
11            long currentTime = clock.getTime();
12     
13            while (iterator.hasNext()) {
14              Map.Entry<O, Long> entry = iterator.next();
15              if (currentTime > entry.getValue() + expireInterval) {
16                iterator.remove();
17                expire(entry.getKey()); // 调用抽象方法expire，会在子类中实现
18                LOG.info("Expired:" + entry.getKey().toString() +
19                        " Timed out after " + expireInterval/1000 + " secs");
20              }
21            }
22          }
23          try {
24            Thread.sleep(monitorInterval);
25          } catch (InterruptedException e) {
26            LOG.info(getName() + " thread interrupted");
27            break;
28          }
29        }
30      }
31    }

1    nmLivelinessMonitor = createNMLivelinessMonitor();
2    addService(nmLivelinessMonitor);

1    private NMLivelinessMonitor createNMLivelinessMonitor() {
2      return new NMLivelinessMonitor(this.rmContext
3          .getDispatcher());
4    }

1    rmDispatcher = setupDispatcher();
2    addIfService(rmDispatcher);
3    rmContext.setDispatcher(rmDispatcher);

1    private Dispatcher setupDispatcher() {
2      Dispatcher dispatcher = createDispatcher();
3      dispatcher.register(RMFatalEventType.class,
4          new ResourceManager.RMFatalEventDispatcher());
5      return dispatcher;
6    }

1    protected Dispatcher createDispatcher() {
2      return new AsyncDispatcher();
3    }

1    @Override
2    public EventHandler getEventHandler() {
3      if (handlerInstance == null) {
4        handlerInstance = new GenericEventHandler();
5      }
6      return handlerInstance;
7    }

01    class GenericEventHandler implements EventHandler<Event> {
02      public void handle(Event event) {
03        if (blockNewEvents) {
04          return;
05        }
06        drained = false;
07     
08        /* all this method does is enqueue all the events onto the queue */
09        int qSize = eventQueue.size();
10        if (qSize !=0 && qSize %1000 == 0) {
11          LOG.info("Size of event-queue is " + qSize);
12        }
13        int remCapacity = eventQueue.remainingCapacity();
14        if (remCapacity < 1000) {
15          LOG.warn("Very low remaining capacity in the event-queue: "
16              + remCapacity);
17        }
18        try {
19          eventQueue.put(event); // 将Event放入到队列eventQueue中
20        } catch (InterruptedException e) {
21          if (!stopped) {
22            LOG.warn("AsyncDispatcher thread interrupted", e);
23          }
24          throw new YarnRuntimeException(e);
25        }
26      };
27    }

01    Runnable createThread() {
02      return new Runnable() {
03        @Override
04        public void run() {
05          while (!stopped && !Thread.currentThread().isInterrupted()) {
06            drained = eventQueue.isEmpty();
07            // blockNewEvents is only set when dispatcher is draining to stop,
08            // adding this check is to avoid the overhead of acquiring the lock
09            // and calling notify every time in the normal run of the loop.
10            if (blockNewEvents) {
11              synchronized (waitForDrained) {
12                if (drained) {
13                  waitForDrained.notify();
14                }
15              }
16            }
17            Event event;
18            try {
19              event = eventQueue.take(); // 从队列取出事件Event
20            } catch(InterruptedException ie) {
21              if (!stopped) {
22                LOG.warn("AsyncDispatcher thread interrupted", ie);
23              }
24              return;
25            }
26            if (event != null) {
27              dispatch(event); // 分发处理该有效事件Event
28            }
29          }
30        }
31      };
32    }

    // Register event handler for RmNodes
2    rmDispatcher.register(
3        RMNodeEventType.class, new NodeEventDispatcher(rmContext)); // 注册：事件类型RMNodeEventType，EventHandler实现类NodeEventDispatcher

01    @Private
02    public static final class NodeEventDispatcher implements
03        EventHandler<RMNodeEvent> {
04     
05      private final RMContext rmContext;
06     
07      public NodeEventDispatcher(RMContext rmContext) {
08        this.rmContext = rmContext;
09      }
10     
11      @Override
12      public void handle(RMNodeEvent event) {
13        NodeId nodeId = event.getNodeId();
14        RMNode node = this.rmContext.getRMNodes().get(nodeId); // 调用getRMNodes()获取到一个ConcurrentMap<NodeId, RMNode>，它维护每个NodeId的状态（RMNode是一个状态机对象）
15        if (node != null) {
16          try {
17            ((EventHandler<RMNodeEvent>) node).handle(event); // RMNode的实现为RMNodeImpl，它也是一个EventHandler
18          } catch (Throwable t) {
19            LOG.error("Error in handling event type " + event.getType()
20                + " for node " + nodeId, t);
21          }
22        }
23      }
24    }

1      private static final StateMachineFactory<RMNodeImpl,
02                                               NodeState,
03                                               RMNodeEventType,
04                                               RMNodeEvent> stateMachineFactory
05                     = new StateMachineFactory<RMNodeImpl,
06                                               NodeState,
07                                               RMNodeEventType,
08                                               RMNodeEvent>(NodeState.NEW)
09    ...
10         .addTransition(NodeState.RUNNING, NodeState.LOST,
11             RMNodeEventType.EXPIRE,
12             new DeactivateNodeTransition(NodeState.LOST))
13    ...
14         .addTransition(NodeState.UNHEALTHY, NodeState.LOST,
15             RMNodeEventType.EXPIRE,
16             new DeactivateNodeTransition(NodeState.LOST))