HBase是一个复杂的分布式非结构化数据库，它将表中的数据按照行的方向切分成一个个的Region，并在若干RegionServer上上线，依靠所在RegionServer对外提供数据读写IO服务。一开始，表中数据由于很少，只有一个Region。随着数据越来越多，一个Region已难以满足频繁的数据读写请求，所以，Region开始分裂。分裂后的两个Region又会按照一定策略选择RegionServer上线，继续对外提供数据读写服务。并且，HBase作为一个分布式数据库，肯定需要考虑负载均衡，它会按照某些策略选择若干Region，在比较繁忙的RegionServer上下线，转移到较为空闲的RegionSever上线继续提供高质量的数据读写服务。所有涉及到的这些Region的上线、下线、分裂，以及我们还没提到的合并等等流程，在HBase内部都是通过不同组件之间发送事件，然后按照一定策略调度执行的。这就是HBase的事件处理模型。
那么，HBase的事件处理模型是如何实现的呢？本文，我们就将研究下HBase内部事件处理模型的实现。
在HBase中有一个抽象类EventHandler，定义如下：
@InterfaceAudience.Private
public abstract class EventHandler implements Runnable, Comparable<Runnable> {
它实现了Runnable接口，说明其子类是一个线程，而且，在它内部定义了以下成员变量：
// type of event this object represents // 该对象代表的事件类型 protected EventType eventType; // 服务器 protected Server server; // sequence id generator for default FIFO ordering of events // 默认的FIFO调度的事件的序列化ID生成器 protected static final AtomicLong seqids = new AtomicLong(0); // sequence id for this event // 该事件的序列化ID private final long seqid; // Listener to call pre- and post- processing. May be null. // 监听器，可能为空 private EventHandlerListener listener; // Time to wait for events to happen, should be kept short // 等待事件发生的时间 protected int waitingTimeForEvents; // 祖先 private final Span parent;
监听器listener是实现了EventHandlerListener接口的实例，接口定义如下：
/**
* This interface provides pre- and post-process hooks for events.
* 为事件提供事前和事后处理钩子的接口
*/
public interface EventHandlerListener {
/**
* Called before any event is processed
* 任何事件执行前被调用
* @param event The event handler whose process method is about to be called.
*/
void beforeProcess(EventHandler event);
/**
* Called after any event is processed
* 任何事件执行后被调用
* @param event The event handler whose process method is about to be called.
*/
void afterProcess(EventHandler event);
}
接口就定义了两个方法，一个是任何事件执行前被调用的beforeProcess()方法和任何事件执行后被调用的afterProcess()方法。
抽象类EventHandler既然实现了Runnable接口，那么其子类肯定是一个线程，而且其功能的实现，必然在核心方法run()方法内。下面，我们就看下这个run()方法，代码如下：
/**
* 线程实现功能的主方法，run()方法，是不是很像一个模板方法啊
*/
public void run() {
<span style="white-space:pre"> </span>
<span style="white-space:pre"> </span>// 开启一个TraceScope
TraceScope chunk = Trace.startSpan(this.getClass().getSimpleName(), parent);
try {
// 先执行监听器的beforeProcess()方法
if (getListener() != null) getListener().beforeProcess(this);
// 接着执行process()方法
process();
// 最后执行监听器的afterProcess()方法
if (getListener() != null) getListener().afterProcess(this);
} catch(Throwable t) {
<span style="white-space:pre"> </span>
// 处理事件异常t
handleException(t);
} finally {
// 关闭TraceScope
chunk.close();
}
}
第一感觉是不是个模板方法呢？它的主要流程就是：
1、开启一个TraceScope;
2、先执行监听器的beforeProcess()方法；
3、接着执行process()方法；
4、最后执行监听器的afterProcess()方法；
5、关闭TraceScope。
如果中间出现Throwable异常，则调用handleException()方法处理事件异常。
关于监听器的beforeProcess()方法和afterProcess()方法我们在上面已经提到过了，这里不再赘述。关键是看一下process()方法，其定义如下：
/** * This method is the main processing loop to be implemented by the various * subclasses. * * 核心方法process()方法是一个抽象方法，子类必须实现 * @throws IOException */ public abstract void process() throws IOException;
它是一个抽象方法，也就意味着子类必须要实现它。并且，它是子类完成事件处理核心逻辑所必须执行的方法。
另外，还实现了诸如获取监听器、设置监听器、获取优先级、获取事件序列ID等工具方法，十分简单，不再一一介绍，代码粘贴如下，读者可自行查看：
/**
* This method is the main processing loop to be implemented by the various
* subclasses.
*
* 核心方法process()方法是一个抽象方法，子类必须实现
* @throws IOException
*/
public abstract void process() throws IOException;
/**
* Return the event type
* 获取时事件类型
* @return The event type.
*/
public EventType getEventType() {
return this.eventType;
}
/**
* Get the priority level for this handler instance. This uses natural
* ordering so lower numbers are higher priority.
* 获取handler实例的优先级，数字越低级别越高
*
* <p>
* Lowest priority is Integer.MAX_VALUE. Highest priority is 0.
* <p>
* Subclasses should override this method to allow prioritizing handlers.
* <p>
* Handlers with the same priority are handled in FIFO order.
* <p>
* @return Integer.MAX_VALUE by default, override to set higher priorities
*/
public int getPriority() {
return Integer.MAX_VALUE;
}
/**
* 获取事件的序列号ID
* @return This events' sequence id.
*/
public long getSeqid() {
return this.seqid;
}
/**
* Default prioritized runnable comparator which implements a FIFO ordering.
* <p>
* Subclasses should not override this. Instead, if they want to implement
* priority beyond FIFO, they should override {@link #getPriority()}.
*
* 实现可比较接口的compareTo()方法，先比较优先级Priority，谁的优先级越小谁就越小。
* 优先级相同的话，再比较事件序列号ID，谁的事件序列号ID越小谁就越小
*/
@Override
public int compareTo(Runnable o) {
EventHandler eh = (EventHandler)o;
if(getPriority() != eh.getPriority()) {
return (getPriority() < eh.getPriority()) ? -1 : 1;
}
return (this.seqid < eh.seqid) ? -1 : 1;
}
/**
* 获取事件的监听器
* @return Current listener or null if none set.
*/
public synchronized EventHandlerListener getListener() {
return listener;
}
/**
* 设置事件的监听器
* @param listener Listener to call pre- and post- {@link #process()}.
*/
public synchronized void setListener(EventHandlerListener listener) {
this.listener = listener;
}
@Override
public String toString() {
return "Event #" + getSeqid() +
" of type " + eventType +
" (" + getInformativeName() + ")";
}
/**
* Event implementations should override thie class to provide an
* informative name about what event they are handling. For example,
* event-specific information such as which region or server is
* being processed should be included if possible.
*/
public String getInformativeName() {
return this.getClass().toString();
}
/**
* 处理事件异常，可能被覆写
* Event exception handler, may be overridden
* @param t Throwable object
*/
protected void handleException(Throwable t) {
LOG.error("Caught throwable while processing event " + eventType, t);
}
接下来就有一个问题，继承了抽象类EventHandler的各种事件是如何被提交的？它们被提交到哪里，又是如何被调度执行的呢？别慌，下面我一一为大家解答。
首先在HRegionServer上有一个叫做service的成员变量，定义如下：
// Instance of the hbase executor service. // HBase执行服务的实例 protected ExecutorService service;
它是HRegionServer上执行各种事件的ExecutorService实例，而ExecutorService提供了通用的事件执行机制，它抽象了线程池、队列，EventType可以被提交，使用线程处理被添加到队列中的对象。如果要创建一个的服务， 创建该类的一个实例，并调用实例的startExecutorService()方法。当服务完成后，调用shutdown()方法。
那么事件是如何被提交的呢？我们以Region上线为例，在HRegionServer对外提供RPC服务的RSRpcServices类的openRegion()方法中，Region上线事件OpenRegionHandler是通过以下方式被提交的，代码如下：
// If there is no action in progress, we can submit a specific handler.
// Need to pass the expected version in the constructor.
// 如果对应Region上没有相关的操作在进行，我们可以提交一个特定的处理者
if (region.isMetaRegion()) {
regionServer.service.submit(new OpenMetaHandler(
regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
} else {
regionServer.updateRegionFavoredNodesMapping(region.getEncodedName(),
regionOpenInfo.getFavoredNodesList());
regionServer.service.submit(new OpenRegionHandler(
regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
}
它就是通过HRegionServer中成员变量service的submit()方法，来提交OpenRegionHandler事件的。