分布式开源任务调度框架TBSchedule深度解析与应用实践

一、下载TBSchedule 源码

http://code.taobao.org/svn/tbschedule/

二、编译TBSchedule源码形成jar包

mvn package/直接mvn deploye/install 一步完成

三、安装到本地的maven仓库

mvn deploy/install

四、在项目pom配置文件中引用这个依赖文件

com.taobao.pamirs.schedule

tbschedule

3.3.3.2

五、配置TBSchedule依赖的zookeeper配置&＃xff0c;TBSchedule的调度依赖于zookeeper的节点配置和心跳链接。

# zooker service address

schedule.zookeeper.address&＃61;120.25.87.176:2181

# root path

schedule.root.path&＃61;/bbb/dd

# session timeout

schedule.session.timeout&＃61;60000

# userName

schedule.zookeeper.username&＃61;ScheduleAdmin

# password

schedule.zookeeper.password&＃61;password

init-method&＃61;"init">

//在控制台连接zookeeper的路径要和该配置的路径一致&＃xff0c;由于配置时默认要检查父节点是否为空&＃xff0c;所以最好先检查该路径是否存在。

六、配置任务项和任务策略&＃xff0c;将任务分片&＃xff0c;查询任务

六-一、通过代码实现任务的分配

private Logger log &＃61; LoggerFactory

.getLogger(getClass());

TBScheduleManagerFactory scheduleManagerFactory;

/**

* 心跳连接时间

*/

private int heartBeatRate &＃61; 5*1000;

private int judgeDeadInterval &＃61; 1*60*1000;

private String taskParameter&＃61;"";

private String strategyTaskParameter&＃61;"";

/**

* jvm最大单线程数量

*/

private int numOfSingleServer &＃61; 2;

/**

* 最大线程组数量 总 最大线程数量 &＃61; jvm最大单线程数量 * 最大的线程组总量

* 将任务列表分成几个队列来处理

*/

private int assignNum &＃61; 2;

private String[] iPLists&＃61;{"127.0.0.1"};

private String[] taskItems&＃61;{"0","1","2","3","4","5","6","7","8","9"};

private String[] baseTaskTypeNames;

private String[] dealBeanNames;

/**

* 开始时间

*/

private String permitRunStartTime;

/**

* 结束时间

*/

private String permitRunEndTime;

/**

* 批处理时 每次处理任务的数量

*/

private int executeNumber &＃61; 1;

/**

* 处理完一批数据后的休息时间

*/

private int sleepTimeInterval &＃61; 0;

/**

* 采集不到数据的休眠时间

*/

private int sleepTimeNoData &＃61; 500;

/**

* 每次采集任务的数量

*/

private int fetchDataNumber &＃61; 500;

&＃64;Autowired

public void setScheduleManagerFactory(

TBScheduleManagerFactory tbScheduleManagerFactory) {

this.scheduleManagerFactory &＃61; tbScheduleManagerFactory;

}

&＃64;Test

public void initialConfigData() throws Exception {

}

&＃64;Override

public void afterPropertiesSet() throws Exception {

if(baseTaskTypeNames.length !&＃61; dealBeanNames.length) {

throw new RuntimeException("task definition error, baseTaskTypeNames length not equals dealBeanNames length");

}

String baseTaskTypeName;

String dealBeanName;

for (int i &＃61; 0; i

baseTaskTypeName &＃61; baseTaskTypeNames[i];

dealBeanName&＃61; dealBeanNames[i];

configTasks(baseTaskTypeName, dealBeanName);

}

}

private void configTasks(String baseTaskTypeName, String dealBeanName)

throws Exception, InterruptedException {

while(this.scheduleManagerFactory.isZookeeperInitialSucess() &＃61;&＃61; false){

Thread.sleep(1000);

}

scheduleManagerFactory.stopServer(null);

Thread.sleep(1000);

try {

this.scheduleManagerFactory.getScheduleDataManager()

.deleteTaskType(baseTaskTypeName);

} catch (Exception e) {

}

ScheduleTaskType baseTaskType &＃61; new ScheduleTaskType();

baseTaskType.setBaseTaskType(baseTaskTypeName);

baseTaskType.setDealBeanName(dealBeanName);

baseTaskType.setHeartBeatRate(heartBeatRate);

baseTaskType.setJudgeDeadInterval(judgeDeadInterval);

baseTaskType.setTaskParameter(taskParameter);

baseTaskType.setTaskItems(taskItems);

baseTaskType.setPermitRunStartTime(permitRunStartTime);

baseTaskType.setPermitRunEndTime(permitRunEndTime);

baseTaskType.setExecuteNumber(executeNumber);

baseTaskType.setSleepTimeInterval(sleepTimeInterval);

baseTaskType.setSleepTimeNoData(sleepTimeNoData);

baseTaskType.setFetchDataNumber(fetchDataNumber);

this.scheduleManagerFactory.getScheduleDataManager()

.createBaseTaskType(baseTaskType);

String taskName &＃61; baseTaskTypeName &＃43; "$TEST";

String strategyName &＃61; baseTaskTypeName &＃43;"-Strategy";

try {

this.scheduleManagerFactory.getScheduleStrategyManager()

.deleteMachineStrategy(strategyName,true);

} catch (Exception e) {

e.printStackTrace();

}

ScheduleStrategy strategy &＃61; new ScheduleStrategy();

strategy.setStrategyName(strategyName);

strategy.setKind(ScheduleStrategy.Kind.Schedule);

strategy.setTaskName(taskName);

strategy.setTaskParameter(strategyTaskParameter);

strategy.setNumOfSingleServer(numOfSingleServer);

strategy.setAssignNum(assignNum);

strategy.setIPList(iPLists);

this.scheduleManagerFactory.getScheduleStrategyManager()

.createScheduleStrategy(strategy);

log.info("创建调度任务成功" &＃43; strategy.toString());

}

六-二、通过TBSchedule自带的控制台来实现任务的配置

七、任务的理解

1、同一个jvm中&＃xff0c;不同线程之间如何防止任务被重复执行&＃xff1f;一个scheduleServer的内部线程间如何进行任务分片&＃xff1f;

答复&＃xff1a;1、数据分片是在不同的jvm&＃xff0c;获知同一个jvm中不同的线程组间起作用。在同一个线程组内的10个线程&＃xff0c;是通过一个同步的任务队列来实现的。2、每个线程从队列中取任务执行&＃xff0c;如果没有任务了&＃xff0c;则由一个线程负责调用selectTasks方法再获取一批新的任务。

主要是设置休眠时间&＃xff1a;即selectTasks方法返回列表的size为0后&＃xff0c;进入休眠。休眠完成之后重新执行该定时任务

2、任务项设置的意义和selectTasks方法的参数含义

答复&＃xff1a;1、 任务项&＃xff08;0,1,2,3,4,5,6,7,8,9&＃xff09;就是任务分片的策略。这个配置就是把数据分成10片。可以表示 ID的最后一位&＃xff0c;也可以是一个独立的字段。根据你的业务来定。

2、 如果只有1组线程&＃xff0c;则所有的任务片都分配给他。这时selectTasks方法的参数&＃xff1a;taskItemNum &＃61;10, queryCondition由10个元素&＃xff0c;分别对应0,1,2,3,4,5,6,7,8,9。

a) 如果只有2组线程&＃xff0c;则任务片被分成两份。这时

b) 一个线程组的selectTasks方法的参数&＃xff1a;taskItemNum &＃61;10, queryCondition有5个元素&＃xff08;0,2,4,6,8&＃xff09;

3、 另外一个线程组的selectTasks方法的参数&＃xff1a;taskItemNum &＃61;10, queryCondition有5个元素&＃xff08;1,3,5,7,9&＃xff09;

4、 如果有10个线程组。则每组线程只会获取到1个任务片。这时selectTasks方法的参数&＃xff1a;taskItemNum &＃61;10, queryCondition只有一个元素&＃xff0c;对应0到9中的一个。

1、 执行期间和时间的修改功能 a) 在创建任务和修改任务的时候&＃xff0c;有两个属性(执行开始时间&＃xff0c;执行结算时间)用于控制任务的执行时间。 b) 时间格式遵循标准的cron格式 http://dogstar.javaeye.com/blog/116130 还增强了原来不支持的倒数第几天的能力。 c) 当时间到底开始时间的时候&＃xff0c;就开始执行任务&＃xff0c;到达结束时间则终止调度&＃xff08;不管是否所有的任务都处理完&＃xff09;。如果没有设置执行结束时间。则一直运行&＃xff0c;直到selectTasks返回的记录数为0&＃xff0c;就终止执行。等待下个开始运行时间在启动。 d) 如果要动态修改任务的执行时间区间&＃xff0c;则先 点击“暂停”按钮&＃xff0c;等所有的服务器都停止完毕&＃xff08;大概需要几秒时间&＃xff09;。当再次单击任务&＃xff0c;出现如下情形表示停止完毕。然后修改执行开始时间&＃xff0c;执行结算时间。在恢复任务调度&＃xff0c;就可以实现调度时间的修改 2、 任务处理的问题 a) Schedule主要是提供任务调度的分配管理。每一个任务是否执行成功&＃xff0c;是通过业务方的bean来实现的。 b) 你需求的例子&＃xff0c;我理解的解决方案如下&＃xff1a; i. 你从云梯拉下来100万数据放到保险应用的数据库中。这个表中有两个关键字段USER_ID和STS(状态 0-未发送&＃xff0c;1&＃xff0d;已发送) ii. 在bean的selectTasks方法的查询sql中除了根据任务进行分片外&＃xff0c;还需要增加状态条件。例如 USER_ID % 10 in( ?,?,?) AND sts &＃61;0 iii. 在bean的execute方法中&＃xff0c;在发送完消息后&＃xff0c;你还需要 修改数据状态 update table STS &＃61;1 where USER_ID &＃61;? 。这样下次就不会取到这条数据了。 iv. 这样就可以保障机器重新启动后&＃xff0c;也不会出现问题。你可以参考DBDemoSingle.java的实现模式。你使用的接口应该是IScheduleTaskDealSingle。 如果旺旺的接口支持批量发送消息的时候&＃xff0c;你才需要使用IScheduleTaskDealMulti接口。