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

yespon 2017-07-13

展开全文

主要内容：
第一部分 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./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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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]1
2
3
4
1
2
3
4

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

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

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

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"}1
2
3
4
5
1
2
3
4
5

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":""}1
2
3
4
5
1
2
3
4
5

同时会在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=true1
2
3
4
5
6
7
8
1
2
3
4
5
6
7
8

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();
        }
    }
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

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

this.zkManager = new ZKManager(p);
...
initialThread = new InitialThread(this);
initialThread.start();1
2
3
4
1
2
3
4

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

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);
        }
    }
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

上述几个节点创建完成，并向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 + "修改成功！！！！";          
}1
2
3
4
5
6
7
8
9
1
2
3
4
5
6
7
8
9

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

//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)删除");
    }
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

如果是更新的话，就不会再创建任务永久节点了，直接修改任务节点内容即可。

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;
}1
2
3
4
5
6
7
8
9
1
2
3
4
5
6
7
8
9

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

//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)删除");
    }
}1
2
3
4
5
6
7
8
9
10
11
12
13
1
2
3
4
5
6
7
8
9
10
11
12
13

如果是更新的话，就不会再创建任务永久节点了，直接修改任务节点内容即可。

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;
} 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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

//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);      
}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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

4. 与其他开源调度框架对比

1）Quartz：Java事实上的定时任务标准。但Quartz关注点在于定时任务而非数据，并无一套根据数据处理而定制化的流程。虽然Quartz可以基于数据库实现作业的高可用，缺少分布式并行执行作业的功能。

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

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

4）TBSchedule：淘宝早期开源，稳定性可以保证。

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

1. TBSchedule源码下载

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

2. 引入源码Demo开发示例

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

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

任务工程依赖

<dependency>
    <groupId>com.taobao.pamirs.schedule</groupId>
    <artifactId>tbschedule</artifactId>
    <version>3.3.3.2</version>
</dependency>1
2
3
4
5
1
2
3
4
5

调度任务实现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;

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

其中selectTasks方法获取需要处理的列表（用集合装着），循环集合中的元素并调用execute方法执行。子计时任务启动，会直到获取不到数据后才停止等待下一个子计时开始，参数后面详细介绍。

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

<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www./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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

环境属性配置文件增加如下配置：

#注册中心地址
schedule.zookeeper.address=172.26.50.86:2181
#定时任务根目录，任意指定，调度控制台配置时对应
schedule.root.catalog=/tbschedule/example
#账户，任意指定，调度控制台配置时对应
schedule.username=username
#密码，任意指定，调度控制台配置时对应
schedule.password=password
#超时配置
schedule.timeout=600001
2
3
4
5
6
7
8
9
10
1
2
3
4
5
6
7
8
9
10

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

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