Yarn Service框架分析
Posted 有山先生
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Yarn Service框架分析相关的知识，希望对你有一定的参考价值。
1. 前言
Yarn是Hadoop最常用的资源调度组件。它的核心设计理念分别是服务化(Service)和状态机(StateMachine)。其中，Service就等价于服务化。本文将介绍Yarn Servcie框架，了解Service框架的设计思路。
2. 什么是服务化
服务化是指将所有Yarn组件提供的服务统一实现Service接口，用具体的Service接口表示抽象的服务概念。如下图所示，Serive接口内部结构如下：
2.1 Service状态表示
Service接口有一个内部枚举类STATE，他表示每个Service的状态可能为4种：NOTINITED(已构建但尚未初始化)、INITED(已初始化但还没有开始或结束)、STARTED(已初始化但还没有开始或结束)、STOPPED(已初始化但还没有开始或结束)，如下所示：
public enum STATE
/** 已构建但尚未初始化 */
NOTINITED(0, "NOTINITED"),
/** 已初始化但还没有开始或结束 */
INITED(1, "INITED"),
/** 已开始，尚未结束 */
STARTED(2, "STARTED"),
/** 已结束，不允许再过度到其他状态 */
STOPPED(3, "STOPPED");
// 一个int值，用来在数组查找和JXM接口。
// 虽然Enamu的ordinal()方法有这个作用，但是随着时间推移提供更多的稳定性保证
private final int value;
private final String statename;
// 状态枚举类的构造方法，跟前文定义的状态序号和状态名匹配
private STATE(int value, String name)
this.value = value;
this.statename = name;
public int getValue()
return value;
@Override
public String toString()
return statename;
2.2 Service关键方法
每个Service都有自身的初始化、启动、停止操作，对应init()、start()、stop()方法。在接口的实现类中：
- init方法需要将Service状态从 NOINITED -> INITED。
- start方法需要将Service状态从 INITED -> STARTED。
- stop方法需要将Service状态从 STARTED -> STOPPED。
上述方法在状态转换过程中，如果发生异常，统一将状态转化为STOPPED。其接口定义如下：
void init(Configuration config);
void start();
void stop();2.3 服务化概念总结
Yarn提供的每个服务都有一种状态(STATE)，可对服务进行初始化/启动/停止操作。
3. 为什么要服务化
Yarn组件对外提供了大量的服务，这些服务来源自不同的开发者，每位开发者的编码习惯都不一样，这样不同服务提供的接口可能大不相同，Yarn管理起来会非常麻烦。如果所有的服务都继承Service接口，每个服务的状态都是NOTINITED、INITED、STARTED、STOPPED之一；且服务都是通过同样的init()、start()、stop()方法启/停，那么Yarn的服务管理会非常清晰。
4. 服务化框架分析：AbstractService
为了方便开发者开发Service服务，Yarn使用抽象类AbstractService实现Service接口的init()、start()、stop()等方法。AbstractService在这些方法中完成了Service状态的转换，这样开发者之需要写具体的服务逻辑即可，不需要重复写状态转换代码。
4.1 AbstractService设计技巧
AbstractService类使用了模版方法设计模式，将状态转化的代码写到init()、start()、stop()中，AbstractService实现类只需要实现类其提供的serviceInit、serviceStart、serviceStop方法，填充服务逻辑即可，帮助开发者节省重复写状态转换代码的时间。
4.1.1 AbstractService#init设计方式
init方法中实现了enterState(STATE.INITED)，使服务从NOINITED -> INITED进行状态转换。init方法内部调用serviceInit方法，serviceInit方法是AbstractService实现类中服务初始化逻辑的方法。
public void init(Configuration conf)
if (conf == null)
throw new ServiceStateException("Cannot initialize service "
+ getName() + ": null configuration");
if (isInState(STATE.INITED))
return;
synchronized (stateChangeLock)
//状态转换NOINITED -> INITED
if (enterState(STATE.INITED) != STATE.INITED)
setConfig(conf);
try
//AbstractService实现类中服务初始化逻辑
serviceInit(config);
if (isInState(STATE.INITED))
//if the service ended up here during init,
//notify the listeners
notifyListeners();
catch (Exception e)
noteFailure(e);
ServiceOperations.stopQuietly(LOG, this);
throw ServiceStateException.convert(e);
抽象类AbstractService提供空的serviceInit的实现，如果AbstractService实现类没有重写serviceInit方法，默认会调用AbstractService#serviceInit方法。
protected void serviceInit(Configuration conf) throws Exception
if (conf != config)
LOG.debug("Config has been overridden during init");
setConfig(conf);
4.1.2 AbstractService#start设计方式
start方法中实现了enterState(STATE.STARTED)，使服务从INITED -> STARTED进行状态转换。start方法内部调用serviceStart方法，serviceStart方法是AbstractService实现类中服务启动的逻辑的方法。
public void start()
if (isInState(STATE.STARTED))
return;
//enter the started state
synchronized (stateChangeLock)
//状态转换INITED -> STARTED
if (stateModel.enterState(STATE.STARTED) != STATE.STARTED)
try
startTime = System.currentTimeMillis();
//AbstractService实现类中服务启动逻辑
serviceStart();
if (isInState(STATE.STARTED))
//if the service started (and isnt now in a later state), notify
LOG.debug("Service is started", getName());
notifyListeners();
catch (Exception e)
noteFailure(e);
ServiceOperations.stopQuietly(LOG, this);
throw ServiceStateException.convert(e);
抽象类AbstractService提供空的serviceStart的实现，如果AbstractService实现类没有重写serviceStart方法，默认会调用AbstractService#serviceStart方法。
protected void serviceStart() throws Exception
4.1.3 AbstractService#stop设计方式
stop方法中实现了enterState(STATE.STOPPED)，使服务从STARTED->STOPPED进行状态转换。stop方法内部调用serviceStop方法，serviceStop方法是AbstractService实现类中服务停止的逻辑的方法。
public void stop()
if (isInState(STATE.STOPPED))
return;
synchronized (stateChangeLock)
//状态转换STARTED -> STOPPED
if (enterState(STATE.STOPPED) != STATE.STOPPED)
try
//AbstractService实现类中服务停止逻辑
serviceStop();
catch (Exception e)
//stop-time exceptions are logged if they are the first one,
noteFailure(e);
throw ServiceStateException.convert(e);
finally
//report that the service has terminated
terminationNotification.set(true);
synchronized (terminationNotification)
terminationNotification.notifyAll();
//notify anything listening for events
notifyListeners();
else
//already stopped: note it
LOG.debug("Ignoring re-entrant call to stop()");
抽象类AbstractService提供空的serviceStop的实现，如果AbstractService实现类没有重写serviceStop方法，默认会调用AbstractService#serviceStop方法。
protected void serviceStop() throws Exception
4.2 AbstractService状态转换
在4.1节中提到AbstractService状态转换。init、start、stop方法中，都是通过AbstractService#enterState方法进行状态转换，本节只通过init状态转换过程为例子，研究Service状态转换过程。
4.2.1 AbstractService状态转换分析
AbstractService#enterState(STATE.INITED)进入STATE.INITED状态：
//传入的STATE.INITED状态是newState，即将要转换的新状态
private STATE enterState(STATE newState)
assert stateModel != null : "null state in " + name + " " + this.getClass();
//调用stateModel成员变量的enterState方法进行状态转换
STATE oldState = stateModel.enterState(newState);
if (oldState != newState)
LOG.debug("Service: entered state ", getName(), getServiceState());
recordLifecycleEvent();
return oldState;
stateModel是AbstractService类的成员变量，它的类型是ServiceStateModel:
public abstract class AbstractService implements Service
private final ServiceStateModel stateModel;
一个服务对象包含一个ServiceStateModel成员变量，每个服务的状态由ServiceStateModel中的state成员表示。state状态流向图如下图所示：
为了防止用户随意转换服务的状态，ServiceStateModel定义了一个二维数组statemap，statemap的false表示禁止的状态转换方式；statemap的true表示允许的状态转换方式。ServiceStateModel定义如下：
横轴代表旧状态，纵轴代表新状态。他们对应的值表示能否从旧状态转化成为新状态。例如：statemap[0][1]为true，表示可以从uninited状态转化成为inited；statemap[3][1]为true，表示不能从stopped状态转化成为inited。
4.2.2 ServiceStateModel状态转换分析
ServiceStateModel#enterState先检查服务能否从旧状态转换成新状态，再更新服务的状态：
public synchronized Service.STATE enterState(Service.STATE proposed)
checkStateTransition(name, state, proposed);
Service.STATE oldState = state;
//atomic write of the new state
state = proposed;
return oldState;
ServiceStateModel#checkStateTransition将自身的state成员变量作为就的状态。检查服务能否从旧状态转换成新状态，如果不能转换，直接抛异常：
public static void checkStateTransition(String name,
Service.STATE state,
Service.STATE proposed)
if (!isValidStateTransition(state, proposed))
throw new ServiceStateException(name + " cannot enter state "
+ proposed + " from state " + state);
ServiceStateModel#isValidStateTransition通过statemap中二维数组的值判断能否从旧状态转换成新状态。如果数组的值是true，表示可以进行转换；如果数组的值是false，表示不可以进行转换：
public static boolean isValidStateTransition(Service.STATE current,
Service.STATE proposed)
boolean[] row = statemap[current.getValue()];
return row[proposed.getValue()];
4.2.3状态转换总结
调用AbstractService#enterState(目的状态)即可从旧状态转换为目的状态。其中，ServiceStateModel的二维数组成员变量statemap用于判断转换是否符合规范。
5. 群组服务：CompositeService
AbstractService类只能表示一个独立的Service，并不能表示所有服务。例如ResourceManager本身就是一个服务，但是它包含一系列子服务，例如管理员服务、系统指标发布服务等等。初始化ResourceManager时，需要初始化ResourceManager的所有子服务。启动和停止ResourceManager服务同理。Yarn提供了CompositeService类，用于群组服务场景。
5.1 群组服务定义
CompositeService继承了AbstractService类，意味着CompositeService服务本身会调用AbstractService#init、AbstractService#start、AbstractService#stop进行服务初始化/服务启动/服务停止。CompositeService的子服务由其成员变量serviceList表示。如下所示：
public class CompositeService extends AbstractService
private final List<Service> serviceList = new ArrayList<Service>();
5.2 群组服务初始化流程
由于群组服务本身就是一个服务，初始化群组服务时，会先调用AbstractService#init方法。AbstractService#init方法中，在转换了服务状态后，调用其实现类#serviceInit。对于群组服务CompositeService，会调用CompositeService#serviceInit方法。在CompositeService#serviceInit方法中，会调用每个子服务的init方法。如果子服务也继承于AbstractService，那么这些子服务初始化时，同样会调用AbstractService#init方法，进行状态转换并调用子服务实现的serviceInit执行初始化逻辑。如果子服务还包含子服务，会递归地执行初始化流程...
protected void serviceInit(Configuration conf) throws Exception
//获取serviceList成员变量，即所有子服务
List<Service> services = getServices();
if (LOG.isDebugEnabled())
LOG.debug(getName() + ": initing services, size=" + services.size());
for (Service service : services)
service.init(conf); //通用一个conf配置文件
//CompositeService的super是AbstractService，AbstractService#serviceInit为空代码块，没什么用
super.serviceInit(conf);
5.3 群组服务启动流程
与初始化流程相似，在启动群组服务时，会先调用AbstractService#start方法。AbstractService#start方法中，在转换了服务状态后，调用其实现类#serviceStart。对于群组服务CompositeService，会调用CompositeService#serviceStart方法。在CompositeService#serviceStart方法中，会调用每个子服务的start方法。如果子服务也继承于AbstractService，那么这些子服务启动时，同样会调用AbstractService#start方法，进行状态转换并调用子服务实现的serviceStart执行初始化逻辑。如果子服务还包含子服务，会递归地执行启动流程...
protected void serviceStart() throws Exception
List<Service> services = getServices();
if (LOG.isDebugEnabled())
LOG.debug(getName() + ": starting services, size=" + services.size());
for (Service service : services)
// start the service. If this fails that service
// will be stopped and an exception raised
service.start();
//CompositeService的super是AbstractService，AbstractService#serviceStart为空代码块，没什么用
super.serviceStart();
5.4 群组服务停止流程
与启动流程相似，在停止群组服务时，会先调用AbstractService#stop方法。AbstractService#stop方法中，在转换了服务状态后，调用其实现类#serviceStop。对于群组服务CompositeService，会调用CompositeService#serviceStop方法。在CompositeService#serviceStop方法中，会调用每个子服务的stop方法。如果子服务也继承于AbstractService，那么这些子服务停止时，同样会调用AbstractService#stop方法，进行状态转换并调用子服务实现的serviceStop执行初始化逻辑。如果子服务还包含子服务，会递归地执行停止流程...
protected void serviceStop() throws Exception
//stop all services that were started
int numOfServicesToStop = serviceList.size();
if (LOG.isDebugEnabled())
LOG.debug(getName() + ": stopping services, size=" + numOfServicesToStop);
stop(numOfServicesToStop, STOP_ONLY_STARTED_SERVICES);
//CompositeService的super是AbstractService，AbstractService#serviceStart为空代码块，没什么用
super.serviceStop();
//可以控制停止的服务数量，并且能够控制只停止已经启动的服务
private void stop(int numOfServicesStarted, boolean stopOnlyStartedServices)
// stop in reverse order of start
Exception firstException = null;
List<Service> services = getServices();
//停止所有子Service
for (int i = numOfServicesStarted - 1; i >= 0; i--)
Service service = services.get(i);
if (LOG.isDebugEnabled())
LOG.debug("Stopping service #" + i + ": " + service);
STATE state = service.getServiceState();
//depending on the stop police
//最终调用service.stop()停止service
if (state == STATE.STARTED || (!stopOnlyStartedServices && state == STATE.INITED))
Exception ex = ServiceOperations.stopQuietly(LOG, service);
if (ex != null && firstException == null)
firstException = ex;
//after stopping all services, rethrow the first exception raised
if (firstException != null)
throw ServiceStateException.convert(firstException);
5.5 群组服务总结
对于包含子服务的服务来说，使用CompositeService初始化/启动/停止服务时，会初始化/启动/停止其所有子服务。
总结
Yarn通过Service->AbstractService->CompositeService类的完善，实现了从简单到复杂的服务设计。实现了所有场景下服务的初始化/启动/停止流程。
以上是关于Yarn Service框架分析的主要内容，如果未能解决你的问题，请参考以下文章