tbschedule使用 在线计算器在线使用

一、简介

tbschedule是淘宝开源的，能够让批量任务或变化的任务，被动态的分配到不同主机（可分布式）的jvm，不同的线程组中并行执行。所有任务能够不重复，不遗漏的执行。

二、tbschedule知识点

2.1 tbschedule特点

tbschedule特点如下：

能够让批量任务或变化的任务，被动态的分配到不同主机（可分布式）的jvm，不同的线程组中并行执行。

所有任务能够不重复，不遗漏的执行。

2.2 tbschedule实现

tbschedule的任务、策略等调度数据是存储在zookeeper中的。

tbschedule的执行是基于jdk的Timer和TimerTask实现的。

tbschedule中的任务，是依附于策略而运行的。也就是说，任务定义了要执行的行为，包括任务名称、取数据和数据处理的bean、每次取数的数量、执行的开始与结束时间、任务项等信息，而策略定义了要执行的任务、在哪台机器上执行、所有机器最大线程组，单个机器线程组数等，并控制任务的执行与停止。

三、tbschedule关键类

3.1 ZKManager

ZKManager就是最基本的zookeeper会话管理类，内容包括zookeeper的创建、会话的连接或重连接、关闭会话等。

3.2 TBScheduleManagerFactory

TBScheduleManagerFactory是tbschedule管理类，包含的功能有：

3.2.1 配置zookeeper

配置zookeeper，并创建zookeeper会话,zookeeper的配置信息有：zkConnectString, rootPath, userName, password, zkSessionTimeout, isCheckParentPath

3.2.2 调度任务和调度策略的管理器生成

ScheduleDataManager4ZK，调度任务管理器（对应在zookeeper中的数据），在此进行初始化和生成。

ScheduleStrategyDataManager4ZK，调度策略管理器（对应在zookeeper中的数据），在此进行初始化和生成。

3.2.3 调度服务的重启、停止等

如stopServer(String strategyName)、stopAll()、reStart()等。

3.3 ScheduleServer

任务处理器（可以理解为线程组），由一组线程（n个线程）组成，每个任务处理器有全局唯一的标识符，一般以IP$UUID[例如192.168.1.100$0C78F0C0FA084E54B6665F4D00FA73DC]的形式出现，一个任务类型的数据可以n个任务处理器处理。内分为Sleep模式和NotSleep模式：

3.3.1 sleep模式

sleep模式，当一线程处理完任务，同时从任务池取不到任务时，若其它线程仍工作，则自己休眠，若其它线程已休眠，则新调取需要处理的数据，同时唤醒其它休眠线程处理数据;

3.3.2 NotSleep模式

NotSleep模式，当一线程处理完任务，同时从任务池取不到任务时，则新调取需要处理的数据，同时唤醒其它休眠线程处理数据;

3.4 TaskItem

任务项，也就是将待处理的任务（数据），进行分片划分，

如：可以按数据的id按10取模，这样就将数年数据划分成了0、1、2、3、4、5、6、7、8、9共10个任务项;

也可按数据的首字母分成了A、B、C、D、E、F、G、H、I、J、K、L、M、N、O、P、Q、R、S、T、U、V、W、X、Y、Z供26个任务项。

这个可以根据需要自行定义的。

3.5 TaskDealBean

自定义的任务处理类，需要实现Schedule的接口IScheduleTaskDealMulti（批处理）或者IScheduleTaskDealSingle（单任务处理），内部主要有两个方法，一个是筛选需当前任务处理器处理的数据，另一个是处理已筛选好的数据。

3.6 OwnSign

环境，指定运行环境，如：开发环境、测试环境、预发环境、生产环境。在筛选当前任务处理器需处理的数据时，会传入该参数。

3.7 ScheduleTaskType

任务的配置类，包括运行的线程数（threadNumber）、运行时间，任务项分组、没数据时的休眠时间、每次取数的量等

3.8 ScheduleStrategy

策略的配置类，所有机器的最大线程组数（assignNum），单个jvm的线程数（numOfSingleServer），运行机器（IPList）等信息。

四、tbschedule使用

这里以tbschedule与spring结合为例，介绍使用的开发步骤。其中尽可能多的理解tbschedule和spring后，能理解tbschedule与spring整合的方式其实有多种形式的，在此介绍一种方式 ，任务和策略在spring启动时进行自动注册。

4.1 添加maven依赖

另一种方式是直接下载源码，这种方式更好，因为目前版本的tbschedule是有bug的，如查看zookeeper连接信息时，tbschedule源码是新建已在代码写死的属性，或者读取tomcat中未有的tbschedule配置类，这是不对，应直接读取已有的zookeeper属性，当然直接maven依赖也不影响使用）,spring依赖包在此略过。

<dependency>

<groupId>com.taobao.pamirs.schedule</groupId>

<artifactId>tbschedule</artifactId>

<version>3.2.18</version>

</dependency>

<dependency>

<groupId>org.apache.zookeeper</groupId>

<artifactId>zookeeper</artifactId>

<version>3.4.6</version>

</dependency>

4.2 下载tbschedule源码

下载地址：http://code.taobao.org/p/tbschedule, 将路径下的tbschedule/branches /3.1.0/src/WebRoot/schedule文件夹复制到工程webapp下，里面tbschdule的jsp文件，便于页面查看与控制任务执行。

4.3 编写自定义抽象任务类

编写自定义抽象任务类，内包含调度任务配置类scheduleTaskType和调度策略配置类scheduleStrategy，便于过会对任务进行自动注册任务和策略信息，该类同时可实现接口IScheduleTaskDealSingle，继承该抽象类的任务类实现其中的方法，如AbstractBaseScheduleTask.java：

public abstract class AbstractBaseScheduleTask<T> implements IScheduleTaskDealSingle<T> {

/**

* 调度任务的配置

*/

private ScheduleTaskType scheduleTaskType;

/**

* 调度策略的配置

*/

private ScheduleStrategy scheduleStrategy;

public ScheduleTaskType getScheduleTaskType() {

return scheduleTaskType;

}

public void setScheduleTaskType(ScheduleTaskType scheduleTaskType) {

this.scheduleTaskType = scheduleTaskType;

}

public ScheduleStrategy getScheduleStrategy() {

return scheduleStrategy;

}

public void setScheduleStrategy(ScheduleStrategy scheduleStrategy) {

this.scheduleStrategy = scheduleStrategy;

}

}

4.4 编写启动类

编写启动类，继承TBScheduleManagerFactory类，用于配置zookeeper信息，实现ApplicationListener接口，并实现其内部方法，用于在spring容器启动后，加载调度任务和调度策略的配置信息到zookeeper中，如类SystemTBScheduleManagerFactory.java :

public class SystemTBScheduleManagerFactory extends TBScheduleManagerFactory implements ApplicationListener<ContextRefreshedEvent>{

@Override

public void onApplicationEvent(ContextRefreshedEvent event) { //注册调度任务和调度策略

try {

super.init(); //默认初始化信息

IScheduleDataManager iScheduleDataManager = null;

ScheduleStrategyDataManager4ZK scheduleStrategyDataManager4ZK = null;

int waitSecond = 120; //默认初始化等待时间，最长120秒

while((null == iScheduleDataManager || null == scheduleStrategyDataManager4ZK) && waitSecond>0){

waitSecond--;

TimeUnit.SECONDS.sleep(1); //等待1秒

try{

iScheduleDataManager = super.getScheduleDataManager();//获取调度任务管理器

scheduleStrategyDataManager4ZK = super.getScheduleStrategyManager();//获取调度策略管理器

}catch (Exception e){

}

}

Assert.notNull(iScheduleDataManager,"初始化tbschedule配置信息失败"); //若仍初始化失败，则抛异常

Assert.notNull(scheduleStrategyDataManager4ZK,"初始化tbschedule配置信息失败"); //若仍初始化失败，则抛异常

Map<String,AbstractBaseScheduleTask> taskMap = event.getApplicationContext().getBeansOfType(AbstractBaseScheduleTask.class);

for(Map.Entry<String,AbstractBaseScheduleTask> m : taskMap.entrySet()){

String key = m.getKey();

AbstractBaseScheduleTask task = m.getValue();

ScheduleTaskType taskType = task.getScheduleTaskType();

taskType.setBaseTaskType("task_"+key); //任务类型(任务名称)

taskType.setDealBeanName(key);

ScheduleStrategy scheduleStrategy = task.getScheduleStrategy();

scheduleStrategy.setStrategyName("strategy_"+key); //策略名称

scheduleStrategy.setTaskName(taskType.getBaseTaskType()); //任务名称

scheduleStrategy.setKind(ScheduleStrategy.Kind.Schedule);

iScheduleDataManager.updateBaseTaskType(taskType);

scheduleStrategyDataManager4ZK.updateScheduleStrategy(scheduleStrategy);

}

} catch (Exception e) {

e.printStackTrace();

}

}

}

4.5 编写任务类

编写任务类，继承前面自定义的抽象类AbstractBaseScheduleTask.java，如：

public class SimpleTask extends AbstractBaseScheduleTask<Date> {

/**

* 执行单个任务

* @param task Object

* @param ownSign 当前环境名称

* @throws Exception

*/

public boolean execute(Date task, String ownSign) throws Exception{

System.out.println("dispose task : "+task.getTime()); //当前任务处理器内的线程处理数据

return true;

}

/**

* 根据条件，查询当前调度服务器可处理的任务

* @param taskParameter 任务的自定义参数

* @param ownSign 当前环境名称

* @param taskItemNum 当前任务类型的任务队列数量

* @param taskItemList 当前调度服务器，分配到的可处理队列

* @param eachFetchDataNum 每次获取数据的数量

* @return

* @throws Exception

*/

public List<Date> selectTasks(String taskParameter, String ownSign, int taskItemNum, List<TaskItemDefine> taskItemList, int eachFetchDataNum) throws Exception {

List<Date> dateList = new ArrayList<>();

List<Long> taskIdList = new ArrayList<>();

for(TaskItemDefine t : taskItemList){ //确定当前任务处理器需处理的任务项id

taskIdList.add(Long.valueOf(t.getTaskItemId()));

}

for(int i=0;i<eachFetchDataNum;i++){ // 添加最多指定数量的待处理数据

Date date = new Date(); //生成待处理数据

Long remainder = date.getTime() % taskItemNum ;

if(taskIdList.contains(remainder)){ //根据数据取模，判断当前待处理数据，是否应由当前任务处理器处理

dateList.add(date);

}

TimeUnit.SECONDS.sleep(1);

}

return dateList; //返回当前任务处理器需要处理的数据

}

/**

* 获取任务的比较器,主要在NotSleep模式下需要用到

* @return

*/

public Comparator<Date> getComparator() {

return null;

}

}

6、配置spring文件spring-tbschedule.xml

配置spring文件spring-tbschedule.xml，包含启动类和任务类配置。

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:context="http://www.springframework.org/schema/context"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context.xsd

">

<!--tbschedule管理器初始化（配置zookeeper,注册调度任务和调度策略）-->

<bean id="systemTBScheduleManagerFactory" class="com.dragon.tbscheduleStudy.system.SystemTBScheduleManagerFactory">

<property name="zkConfig">

<map>

<entry key="zkConnectString" value="127.0.0.1:2181" />

<entry key="rootPath" value="/myself/tbscheduleStudy" />

<entry key="userName" value="root" />

<entry key="password" value="123456" />

<entry key="zkSessionTimeout" value="8000" />

<entry key="isCheckParentPath" value="true" />

</map>

</property>

</bean>

<!--任务simpleTask-->

<bean id="simpleTask" class="com.dragon.tbscheduleStudy.task.SimpleTask" >

<property name="scheduleTaskType">

<bean class="com.taobao.pamirs.schedule.taskmanager.ScheduleTaskType">

<!--允许执行的开始时间-->

<property name="permitRunStartTime" value="0 0 0 * * ?" />

<!--允许执行的结束时间-->

<property name="permitRunEndTime" value="59 59 23 * * ?" />

<!--当没有数据的时候，休眠的时间-->

<property name="sleepTimeNoData" value="3000" />

<!--在每次数据处理晚后休眠的时间-->

<property name="sleepTimeInterval" value="1000" />

<!--每次获取数据的数量-->

<property name="fetchDataNumber" value="10" />

<!--任务项数组-->

<property name="taskItems">

<list>

<value>0:{TYPE=A,KIND=1}</value>

<value>1:{TYPE=B,KIND=2}</value>

<value>2:{TYPE=C,KIND=3}</value>

</list>

</property>

</bean>

</property>

<property name="scheduleStrategy">

<bean class="com.taobao.pamirs.schedule.strategy.ScheduleStrategy">

<!--最大线程组数量-->

<property name="assignNum" value="9" />

<!--单个机器的线程组数量-->

<property name="numOfSingleServer" value="3" />

<!--策略运行的机器-->

<property name="IPList">

<list>

<value>127.0.0.1</value>

</list>

</property>

</bean>

</property>

</bean>

</beans>

4.7 配置web.xml文件

配置web.xml文件，如：

<?xml version="1.0" encoding="UTF-8"?>

<web-app xmlns="http://java.sun.com/xml/ns/javaee"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://java.sun.com/xml/ns/javaee

http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"

version="3.0">

<context-param>

<param-name>contextConfigLocation</param-name>

<param-value>classpath:spring-tbschedule.xml</param-value>

</context-param>

<listener>

<listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>

</listener>

</web-app>

4.8 修改项目index.jsp页面

修改项目的index.jsp页面，重定向到tbschedule首页，如：

<%

response.sendRedirect("schedule/index.jsp");

%>

至此，开发流程结束，

启动项目，在浏览器输入http://localhost:8080/schedule/index.jsp，可查看任务情况，

若想修改配置，则输入地址http://localhost:8080/schedule/index.jsp?manager=true即可