分布式开源调度框架TBSchedule原理与应用

主要内容：

第一部分 TBSchedule基本概念及原理

1. 概念介绍

2. 工作原理

3. 源代码分析

4. 与其它开源调度框架对照

第二部分 TBSchedule分布式调度演示样例

1. TBSchedule源代码下载

2. 引入源代码Demo开发演示样例

3. 控制台配置任务调度

4. selectTasks方法參数说明

5. 创建调度策略參数说明

6. 创建任务參数说明

---

第一部分 TBSchedule基本概念及原理

1. 概念介绍

TBSchedule是一个支持分布式的调度框架。能让一种批量任务或者不断变化的任务，被动态的分配到多个主机的JVM中，不同的线程组中并行运行。基于ZooKeeper的纯Java实现，由Alibaba开源。

2. 工作原理

TBSchesule对分布式的支持包含调度机的分布式和运行机的分布式，其网络部署架构图例如以下：

2.1 数据存储

运行机和调度机均以ZooKeeper为注冊中心，全部数据以节点及节点内容的形式注冊，通过定时汇报主机状态保持存活在ZooKeeper上。

首先看下运行机对ZooKeeper的使用配置文件：

<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
<beans default-autowire="byName">
    <bean id="scheduleManagerFactory"
        class="com.taobao.pamirs.schedule.strategy.TBScheduleManagerFactory"
        init-method="init">
        <property name="zkConfig">
            <map>
                <entry key="zkConnectString" value="${schedule.zookeeper.address}" />
                <entry key="rootPath" value="${schedule.root.catalog}" />
                <entry key="zkSessionTimeout" value="${schedule.timeout}" />
                <entry key="userName" value="${schedule.username}" />
                <entry key="password" value="${schedule.password}" />
                <entry key="isCheckParentPath" value="true" />
            </map>
        </property>
    </bean>
</beans>

1）运行机部署启动，会在ZooKeeper上创建永久根节点schedule.zookeeper.address，其后全部的操作均在该根节点下进行。

这里以/ttest/creditjob为根节点。查看运行机注冊后情况：

[zk: 172.26.50.86:2181(CONNECTED) 28] ls /ttest/creditjob
[strategy, baseTaskType, factory]
[zk: 172.26.50.86:2181(CONNECTED) 29] ls /ttest/creditjob/factory
[127.0.0.1$MIE-ZHANGTAO-D1$9D3029EC0C574403B6CFD0C146644A77$0000000000, 127.0.0.1$MIE-ZHANGTAO-D1$D826BC6565DC4D6CB85F7AE321EE51AE$0000000001]

能够看到根节点以下有3个永久子节点，strategy存储调度机创建的策略信息，baseTaskType存储调度机创建的任务信息，factory存储运行机注冊的主机信息。每台运行机启动后。都会在factory下创建一个暂时顺序子节点，该节点名是由TBSchedule源代码生成的主机唯一表示。

根节点内容为当前TBSchedule内置版本号号。可在程序改动，实际没什么意义。

[zk: 172.26.50.86:2181(CONNECTED) 17] get /ttest/creditjob
tbschedule-3.2.12

2）调度机部署启动，这时不会对ZooKeeper节点做不论什么操作。打开调度机配置面板：

配置好ZooKeeper接入点，点击管理主页。进入调度任务管理面板：

输入各项參数创建新任务后，此时会在baseTaskType以下创建任务名称永久子节点（调度机全部都宕机重新启动后。仍能保持数据的完整性）。而当前节点的内容就是配置的各项參数。

[zk: 172.26.50.86:2181(CONNECTED) 37] ls /ttest/creditjob/baseTaskType
[IScheduleTaskDealSingleTest]

[zk: 172.26.50.86:2181(CONNECTED) 39] get /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest
{"baseTaskType":"IScheduleTaskDealSingleTest","heartBeatRate":5000,"judgeDeadInterval":60000,"sleepTimeNoData":500,"sleepTimeInterval":0,"fetchDataNumber":500,"executeNumber":1,"threadNumber":1,"processorType":"SLEEP","permitRunStartTime":"0 * * * * ?","expireOwnSignInterval":1.0,"dealBeanName":"iScheduleTaskDealSingleTest","taskParameter":"0","taskKind":"static","taskItems":["0"],"maxTaskItemsOfOneThreadGroup":0,"version":0,"sts":"resume"}

3）创建调度策略，控制调度机调度状态。

创建完毕调度策略后开启调度，此过程会在相应的任务节点strategy下创建永久子节点并写入策略数据。在该子节点下创建表示调度机的暂时顺序子节点并写入调度策略数据。

[zk: 172.26.50.86:2181(CONNECTED) 56] get /ttest/creditjob/strategy/IScheduleTaskDealSingleTest
{"strategyName":"IScheduleTaskDealSingleTest","IPList":["127.0.0.1"],"numOfSingleServer":0,"assignNum":1,"kind":"Schedule","taskName":"IScheduleTaskDealSingleTest","taskParameter":"0","sts":"resume"}

[zk: 172.26.50.86:2181(CONNECTED) 57] get /ttest/creditjob/strategy/IScheduleTaskDealSingleTest/127.0.0.1$MIE-ZHANGTAO-D1$9D3029EC0C574403B6CFD0C146644A77$0000000000
{"strategyName":"IScheduleTaskDealSingleTest","uuid":"127.0.0.1$MIE-ZHANGTAO-D1$9D3029EC0C574403B6CFD0C146644A77$0000000000","requestNum":1,"currentNum":0,"message":""}

同一时候会在baseTaskType/IScheduleTaskDealSingleTest下创建下创建两层永久子节点并注冊调度主机数据。

[zk: 172.26.50.86:2181(CONNECTED) 45] ls /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest/IScheduleTaskDealSingleTest
[taskItem, server]

[zk: 172.26.50.86:2181(CONNECTED) 50] get /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest/IScheduleTaskDealSingleTest/taskItem
IScheduleTaskDealSingleTest$127.0.0.1$4E8008EE18334564BE1E31C7C0D55296$0000000000,IScheduleTaskDealSingleTest$127.0.0.1$4E8008EE18334564BE1E31C7C0D55296$0000000001

[zk: 172.26.50.86:2181(CONNECTED) 51] get /ttest/creditjob/baseTaskType/IScheduleTaskDealSingleTest/IScheduleTaskDealSingleTest/server  
reload=true

2.2 分布式高可用高效率保障

1）调度机的高可用有保障，多调度机向注冊中心注冊后，共享调度任务。且同一调度任务仅由一台调度机运行调度。当前调度机异常宕机后。其余的调度机会接上。

2）运行机的高可用有保障，多运行机向注冊中心注冊后，配置运行机单线程（多机总线程为1）运行任务。调度机会随机启动一台运行机运行。当前运行异常机宕机后。调度机会会新调度一台运行机。

3）运行机的并行高效保障，配置运行机多线程且划分多任务子项后，各任务子项均衡分配到全部运行机，各运行机均运行，多线程数据一致性协调由任务项參数区分。

4）弹性扩展失效转移保障，运行中的运行机宕机，或新增运行机。调度机将在下次任务运行前又一次分配任务项，不影响正常运行机任务（崩溃的运行机当前任务处理失效）。运行中的调度机宕机或动态新增调度机，不影响运行机当前任务，调度机宕机后动态切换。

3. 源代码分析

3.1 运行机注冊节选

从Spring配置文件能够看到，运行机注冊的入口在TBScheduleManagerFactory的init方法，代码片段：

public class TBScheduleManagerFactory implements ApplicationContextAware {
    public void init() throws Exception {
        Properties properties = new Properties();
        for(Map.Entry<String,String> e: this.zkConfig.entrySet()){
            properties.put(e.getKey(),e.getValue());
        }
        this.init(properties);
    }

    public void init(Properties p) throws Exception {
        if(this.initialThread != null){
            this.initialThread.stopThread();
        }
        this.lock.lock();
        try{
            this.scheduleDataManager = null;
            this.scheduleStrategyManager = null;
            ConsoleManager.setScheduleManagerFactory(this);
            if(this.zkManager != null){
                this.zkManager.close();
            }
            this.zkManager = new ZKManager(p);
            this.errorMessage = "Zookeeper connecting ......" + this.zkManager.getConnectStr();
            initialThread = new InitialThread(this);
            initialThread.setName("TBScheduleManagerFactory-initialThread");
            initialThread.start();
        }finally{
            this.lock.unlock();
        }
    }
}

init方法将配置參数封装到Properties对象后開始初始化，连接上ZooKeeper并启动一个新的线程进行节点数据处理。

this.zkManager = new ZKManager(p);
...
initialThread = new InitialThread(this);
initialThread.start();

跟踪代码能够看到新线程调用的实际处理方法是：

public void initialData() throws Exception{

    /** 递归创建永久根节点（/ttest/creditjob）并写入版本号信息 */
    this.zkManager.initial();

    /** 创建永久子节点 baseTaskType */
    this.scheduleDataManager = new ScheduleDataManager4ZK(this.zkManager);

    /** 创建永久子节点 strategy 和 factory */
    this.scheduleStrategyManager  = new ScheduleStrategyDataManager4ZK(this.zkManager);

    if (this.start == true) {

        /** 注冊调度管理器，创建暂时顺序子节点。节点表示主机的注冊信息 */
        this.scheduleStrategyManager.registerManagerFactory(this);
        if(timer == null){
            timer = new Timer("TBScheduleManagerFactory-Timer");
        }

        if(timerTask == null){
            /** 启动一个定时器检測ZooKeeper状态，假设连接失败，关闭全部的任务后，又一次连接Zookeeper服务器 */
            timerTask = new ManagerFactoryTimerTask(this);
            timer.schedule(timerTask, 2000,this.timerInterval);
        }
    }
}

上述几个节点创建完毕，并向ZooKeeper注冊监听，当有数据变化时获得通知（任务运行/暂停）。到这里。就完毕了运行机到ZooKeeper的注冊监听过程。

3.2 调度任务创建节选

任务创建提交保存为入口。将參数封装到ScheduleTaskType对象中，调用节点创建和更新方法：

//taskTypeEdit.jsp->taskTypeDeal.jsp

if(action.equalsIgnoreCase("createTaskType")){
    ConsoleManager.getScheduleDataManager().createBaseTaskType(taskType);
    result = "任务" + baseTaskType + "创建成功！
！！
！
";
}else{
    ConsoleManager.getScheduleDataManager().updateBaseTaskType(taskType);
    result = "任务" + baseTaskType + "改动成功！
！
！。";          
}

真正运行任务节点及数据处理的代码段：

//ScheduleDataManager4ZK.java

public void createBaseTaskType(ScheduleTaskType baseTaskType) throws Exception {
    if(baseTaskType.getBaseTaskType().indexOf("$") > 0){
        throw new Exception("调度任务" + baseTaskType.getBaseTaskType() +"名称不能包含特殊字符 $");
    }

    /** 在 baseTaskType 节点下创建任务永久节点并写入节点内容为任务配置參数 */
    String zkPath = this.PATH_BaseTaskType + "/"+ baseTaskType.getBaseTaskType();
    String valueString = this.gson.toJson(baseTaskType);
    if ( this.getZooKeeper().exists(zkPath, false) == null) {
        this.getZooKeeper().create(zkPath, valueString.getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
    } else {
    throw new Exception("调度任务" + baseTaskType.getBaseTaskType() + "已经存在,假设确认须要重建，请先调用deleteTaskType(String baseTaskType)删除");
    }
}

假设是更新的话，就不会再创建任务永久节点了，直接改动任务节点内容就可以。

3.3 策略创建节选

策略创建提交保存为入口，将參数封装到ScheduleStrategy对象中。调用节点创建和更新方法：

//scheduleStrategyEdit.jsp->scheduleStrategyDeal.jsp

if (action.equalsIgnoreCase("createScheduleStrategy")) {
    ConsoleManager.getScheduleStrategyManager().createScheduleStrategy(scheduleStrategy);
    isRefreshParent = true;
} else if (action.equalsIgnoreCase("editScheduleStrategy")) {
    ConsoleManager.getScheduleStrategyManager().updateScheduleStrategy(scheduleStrategy);
    isRefreshParent = true;
}

真正运行任务节点及数据处理的代码段：

//ScheduleStrategyDataManager4ZK.java

public void createScheduleStrategy(ScheduleStrategy scheduleStrategy) throws Exception {
    String zkPath = this.PATH_Strategy + "/"+ scheduleStrategy.getStrategyName();

    /** 在 strategy 节点下创建任务永久节点并写入节点内容为任务配置參数 */
    String valueString = this.gson.toJson(scheduleStrategy);
    if ( this.getZooKeeper().exists(zkPath, false) == null) {
        this.getZooKeeper().create(zkPath, valueString.getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
    } else {
        throw new Exception("调度策略" + scheduleStrategy.getStrategyName() + "已经存在,假设确认须要重建，请先调用deleteMachineStrategy(String taskType)删除");
    }
}

假设是更新的话，就不会再创建任务永久节点了，直接改动任务节点内容就可以。

3.4 调度控制节选

策略调度控制代码段：

//scheduleStrategyList.jsp->scheduleStrategyDeal.jsp

else if (action.equalsIgnoreCase("deleteScheduleStrategy")) {
    ConsoleManager.getScheduleStrategyManager()
                    .deleteMachineStrategy(
    scheduleStrategy.getStrategyName());
    isRefreshParent = true;
} else if (action.equalsIgnoreCase("pauseTaskType")) {
    ConsoleManager.getScheduleStrategyManager().pause(
                    scheduleStrategy.getStrategyName());
    isRefreshParent = true;
} else if (action.equalsIgnoreCase("resumeTaskType")) {
    ConsoleManager.getScheduleStrategyManager().resume(
                    scheduleStrategy.getStrategyName());
    isRefreshParent = true;
} 

真正运行任务节点及数据处理的代码段：

//ScheduleStrategyDataManager4ZK.java

/** 策略删除，即删除strategy下相应的策略节点及数据 */
public void deleteMachineStrategy(String taskType) throws Exception {
    deleteMachineStrategy(taskType,false);
}

/** 调度暂停，即改动strategy下相应的策略节点的状态标示数据 */
public void pause(String strategyName) throws Exception{
    ScheduleStrategy strategy = this.loadStrategy(strategyName);
    strategy.setSts(ScheduleStrategy.STS_PAUSE);
    this.updateScheduleStrategy(strategy);
}

/** 调度启动，即改动strategy下相应的策略节点的状态标示数据 */
public void resume(String strategyName) throws Exception{
    ScheduleStrategy strategy = this.loadStrategy(strategyName);
    strategy.setSts(ScheduleStrategy.STS_RESUME);
    this.updateScheduleStrategy(strategy);      
}

改动节点数据，通过ZooKeeper的事件通知机制，让运行机获得变更通知。

4. 与其它开源调度框架对照

1）Quartz：Java其实的定时任务标准。

但Quartz关注点在于定时任务而非数据。并无一套依据数据处理而定制化的流程。

尽管Quartz能够基于数据库实现作业的高可用。缺少分布式并行运行作业的功能。

2）Crontab：Linux系统级的定时任务运行器。缺乏分布式和集中管理功能。

3）elastic-job：当当网近期开源项目，功能跟TBSchedule差点儿一样（批斗TBSchedule文档缺失严重），一台服务器仅仅能开启一个任务实例，基于Ip不基于IpPort，单机难调试集群功能。

4）TBSchedule：淘宝早期开源。稳定性能够保证。

---

第二部分 TBSchedule分布式调度演示样例

1. TBSchedule源代码下载

下载TBSchedule源代码project，内容包含两部分：project编译成jar的任务开发依赖包和project编译成war的调度控制台。

2. 引入源代码Demo开发演示样例

当前演示样例与Spring集成，源代码可作为普通project依赖入任务project，也可将其打包成jar并引入依赖，此处版本号为3.2.2.2。

补充：若打包失败，请检查编译插件版本号及jdk编译版本号。

任务project依赖

<dependency>
    <groupId>com.taobao.pamirs.schedule</groupId>
    <artifactId>tbschedule</artifactId>
    <version>3.3.3.2</version>
</dependency>

调度任务实现IScheduleTaskDealSingle。并实现selectTasks。execute方法，具体演示样例：

Component("iScheduleTaskDealSingleTest")
public class IScheduleTaskDealSingleTest implements IScheduleTaskDealSingle<TaskModel> {

    private static final Logger LOG = LoggerFactory.getLogger(IScheduleTaskDealSingleTest.class);

    @Override
    public Comparator<TaskModel> getComparator() {
        return null;
    }

    @Override
    public List<TaskModel> selectTasks(String taskParameter, String ownSign, int taskQueueNum,
            List<TaskItemDefine> taskItemList, int eachFetchDataNum) throws Exception {

        LOG.info("IScheduleTaskDealSingleTest配置的參数，taskParameter:{}。ownSina:{}。taskQueueNum:{},taskItemList:{}, eachFetchDataNum:{}", taskParameter, ownSign, taskQueueNum, taskItemList, eachFetchDataNum);

        LOG.info("IScheduleTaskDealSingleTest选择任务列表開始啦..........");
        List<TaskModel> models = new ArrayList<TaskModel>();
        models.add(new TaskModel(String.valueOf(System.currentTimeMillis()), "taosirTest1"));
        models.add(new TaskModel(String.valueOf(System.currentTimeMillis()), "taosirTest2"));

        return models;

    }

    @Override
    public boolean execute(TaskModel model, String ownSign) throws Exception {

        LOG.info("IScheduleTaskDealSingleTest运行開始啦.........." + new Date());
        System.out.println(model);
        return true;

    }

}

当中selectTasks方法获取须要处理的列表（用集合装着），循环集合中的元素并调用execute方法运行。子计时任务启动，会直到获取不到数据后才停止等待下一个子计时開始，參数后面具体介绍。

将调度任务注冊到zookeeper中心，spring中引入例如以下配置：

<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
<beans default-autowire="byName">
    <bean id="scheduleManagerFactory"
        class="com.taobao.pamirs.schedule.strategy.TBScheduleManagerFactory"
        init-method="init">
        <property name="zkConfig">
            <map>
                <entry key="zkConnectString" value="${schedule.zookeeper.address}" />
                <entry key="rootPath" value="${schedule.root.catalog}" />
                <entry key="zkSessionTimeout" value="${schedule.timeout}" />
                <entry key="userName" value="${schedule.username}" />
                <entry key="password" value="${schedule.password}" />
                <entry key="isCheckParentPath" value="true" />
            </map>
        </property>
    </bean>
</beans>

环境属性配置文件添加例如以下配置：

#注冊中心地址
schedule.zookeeper.address=172.26.50.86:2181
#定时任务根文件夹。随意指定，调度控制台配置时相应
schedule.root.catalog=/tbschedule/example
#账户，随意指定。调度控制台配置时相应
schedule.username=username
#密码，随意指定，调度控制台配置时相应
schedule.password=password
#超时配置
schedule.timeout=60000

启动容器，iScheduleTaskDealSingleTest就完毕了到zookeeper中心的注冊。

补充：TBSchedule提供了IScheduleTaskDealSingle和IScheduleTaskDealMulti两个接口，个人在測试中发现两者除了execute方法上參数不同外，功能上并没有别的不同，仅仅是语义上的区分，在处理模式为SLEEP下getComparator()没实用。普通情况下，都是SLEEP模式。

3. 控制台配置任务调度

将控制台ScheduleConsole.war部署到tomcat容器。

补充：我通过ant运行源代码中的build.xml构建控制台。部署运行失败（没用过ant，眼下不知道原因），这样的情况下：

使用方式一，直接用下载包中的控制台部署就可以。

使用方式二，改动project配置打成war包，这灵活，还能够自己定义改动，源文件不支持中文，可将编码改成utf-8支持。

向注冊中心注冊配置（跟任务注冊用同一根文件夹，官方wiki图示）
http://{server}:{port}/ScheduleConsole/schedule/config.jsp

配置调度任务（官方wiki图示）
http://{server}:{port}/ScheduleConsole/schedule/index.jsp

4. selectTasks方法參数说明

taskParameter：相应控制台自己定义參数，可自己定义传入做逻辑上的操作

taskQueueNum：相应控制台任务项数量

taskItemList：集合中TaskItemDefine的id值相应任务项值，多线程处理时，依据任务项协调数据一致性和完整性

eachFetchDataNum：相应控制台每次获取数量，因为子计时单元開始后，会不断的去取数据进行处理，直到取不到数据子计时才停止，等待下一个子计时開始。

能够限制每次取数。防止一次性数据记录过大，内存不足。

ownSign：环境參数，一般没什么用

5. 创建调度策略參数说明

策略名称：策略标示，可随意填写

任务类型：一般保持默认Schedule

任务名称：相应任务栏被调度任务名称

任务參数：一般不用，保持默认

单JVM最大线程组数量：单个JVM同意开启的线程数

最大线程组数量：多处理机情况下的线程总数限制（总线程为2。任务项线程为4是没有意义的）

IP地址：127.0.0.1或者localhost会在全部机器上运行，注意多处理机若没有依据任务子项划分数据处理，会导致多处理机反复处理数据，慎重配置

创建演示样例。官方wiki上有图示。上面主要是各參数的具体含义。

6. 创建任务參数说明

任务名称：策略调度的标示。一旦创建保存，不可更改

任务处理的SpringBean：注冊到spring的任务bean，如iScheduleTaskDealSingleTest

心跳频率/假定服务死亡时间/处理模式/没有数据时休眠时长/运行结束时间：一般保持默认就可以

线程数：处理该任务的线程数，在没有划分多任务项的情况下，多线程是没有意义的，且线程数量大于任务项也是没有意义的（线程数小于等于任务项）。注意假设开启多线程，必须对数据做任务项过滤

单线程组最大任务项：配置单JVM处理的最大任务项数量，多任务项情况下。可按需限制，一般默认，多运行机会均衡分配

每次获取数量：子计时单元開始。线程会不断的去获取数据（每次获取的限制）并处理数据，直到获取不到数据子计时才结束（方法内不用就能够随意配置）

每次运行数量：//还没測试过（可能是将获取的数量拆分多次运行）

每次处理完休眠时间：子计时单元開始，仅仅要有数据，就会不停的获取不停的处理，这个时间设置后，子计时单元開始每次获取运行后。无论还有没有待数据，都先歇会儿再获取处理

自己定义參数：可自己定义控制任务逻辑操作

任务项：这项非常重要。在多线程情况下，划分任务项是有意义的，可是要注意必须通过任务项參数，协调待处理数据。否则多线程会反复处理

创建演示样例，官方wiki上有图示。上面主要是各參数的具体含义。