基於zookeeper和quartz實現分散式定時排程
阿新 • • 發佈：2018-12-22
目的
利用zookeeper的特性，來控制quartz實現分散式排程，保證quartz的單點執行，同時解除quartz自身分散式部署對資料庫的依賴，保證同一時刻只有一個quartz應用在執行任務。
實現方式
利用zk的分散式獨佔鎖，控制quartz應用執行節點，讓拿到獨佔鎖的quartz應用執行排程，沒有拿到獨佔鎖的quartz處理等待狀態。
類圖
核心程式碼
public class TriggerBean { /** * 標識 */ private String key; /** * 所屬組 */ private String group; /** * 描述 */ private String description; /** * 啟動時間 */ private String startTime; /** * 結束時間 */ private String endTime; /** * 優先順序 */ private Integer priority; /** * 日曆名稱 */ private String calendarName; /** * 失火指令（引數0,1,2） * MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY = -1 * MISFIRE_INSTRUCTION_SMART_POLICY = 0 (預設) * MISFIRE_INSTRUCTION_FIRE_ONCE_NOW = 1 * MISFIRE_INSTRUCTION_DO_NOTHING = 2 */ private Integer misfireInstruction; /** * 任務代理類 */ private JobDetailProxyBean jobDetail; public String getKey() { return key; } public void setKey(String key) { this.key = key; } public String getGroup() { return group; } public void setGroup(String group) { this.group = group; } public String getDescription() { return description; } public void setDescription(String description) { this.description = description; } public String getStartTime() { return startTime; } public void setStartTime(String startTime) { this.startTime = startTime; } public String getEndTime() { return endTime; } public void setEndTime(String endTime) { this.endTime = endTime; } public Integer getPriority() { return priority; } public void setPriority(Integer priority) { this.priority = priority; } public String getCalendarName() { return calendarName; } public void setCalendarName(String calendarName) { this.calendarName = calendarName; } public Integer getMisfireInstruction() { return misfireInstruction; } public void setMisfireInstruction(Integer misfireInstruction) { this.misfireInstruction = misfireInstruction; } public JobDetailProxyBean getJobDetail() { return jobDetail; } public void setJobDetail(JobDetailProxyBean jobDetail) { this.jobDetail = jobDetail; } }
public class CronTriggerBean extends TriggerBean { /** * CRON表示式 */ private String cronExpression; public String getCronExpression() { return cronExpression; } public void setCronExpression(String cronExpression) { this.cronExpression = cronExpression; } }
public class SimpleTriggerBean extends TriggerBean { /** * 時間間隔(秒) */ private Integer interval; /** * 重複次數(預設：-1為無限迴圈) */ private Integer repeatCount; public Integer getInterval() { return interval; } public void setInterval(Integer interval) { this.interval = interval; } public Integer getRepeatCount() { return repeatCount; } public void setRepeatCount(Integer repeatCount) { this.repeatCount = repeatCount; } }
public class SchedulerFactoryBean implements InitializingBean {
protected static Logger logger = Logger.getLogger(SchedulerFactoryBean.class);
/**
* 觸發器列表
*/
private List<Object> triggers;
/**
* zooKeeper工廠
*/
private ZookeeperFactory zooKeeperFactory;
/**
* Spring初始化方法
* @throws SchedulerException
*/
public void afterPropertiesSet() throws SchedulerException {
this.initSchedulerFactory();
}
/**
* 初始化排程器工廠
* @throws SchedulerException
*/
public void initSchedulerFactory() throws SchedulerException {
//初始化StdSchedulerFactory
StdSchedulerFactory schedulerFactory = SchedulerUtils.initStdSchedulerFactory();
//獲取排程器
Scheduler scheduler = schedulerFactory.getScheduler();
//裝載排程器
for(Object triggerObject : this.getTriggers()){
if(triggerObject instanceof CronTriggerBean){
CronTriggerBean cronTriggerBean = (CronTriggerBean)triggerObject;
//獲取任務代理類物件
JobDetailProxyBean jobDetailProxyBean = cronTriggerBean.getJobDetail();
//裝配任務
JobDetail jobDetail = SchedulerUtils.assemblyJobDetail(jobDetailProxyBean);
//設定zooKeeper連線工廠
jobDetail.getJobDataMap().put("zooKeeperFactory",this.getZooKeeperFactory());
//裝配觸發器
CronTrigger cronTrigger = SchedulerUtils.assemblyCronTrigger(cronTriggerBean);
scheduler.scheduleJob(jobDetail, cronTrigger);
// System.out.println("CronTriggerBean");
}else{
SimpleTriggerBean simpleTriggerBean = (SimpleTriggerBean)triggerObject;
//獲取任務代理類物件
JobDetailProxyBean jobDetailProxyBean = simpleTriggerBean.getJobDetail();
//裝配任務
JobDetail jobDetail = SchedulerUtils.assemblyJobDetail(jobDetailProxyBean);
//設定zooKeeper連線工廠
jobDetail.getJobDataMap().put("zooKeeperFactory",this.getZooKeeperFactory());
//裝配觸發器
SimpleTrigger simpleTrigger = SchedulerUtils.assemblySimpleTrigger(simpleTriggerBean);
scheduler.scheduleJob(jobDetail, simpleTrigger);
// System.out.println("SimpleTriggerBean");
}
}
scheduler.start();
logger.info("排程器已啟動");
}
public List<Object> getTriggers() {
return triggers;
}
public void setTriggers(List<Object> triggers) {
this.triggers = triggers;
}
public ZookeeperFactory getZooKeeperFactory() {
return zooKeeperFactory;
}
public void setZooKeeperFactory(ZookeeperFactory zooKeeperFactory) {
this.zooKeeperFactory = zooKeeperFactory;
}
}
package com.ab.scheduling.quartz;
import com.ab.scheduling.quartz.constant.Constant;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.springframework.beans.factory.InitializingBean;
import java.util.Collections;
import java.util.List;
/**
* Zookeeper 工廠類
* Date: 14-4-2
* Time: 下午4:03
*/
public class ZookeeperFactory implements InitializingBean{
public static Logger logger = Logger.getLogger(ZookeeperFactory.class);
/**
* zookeeper服務地址
*/
private String hosts;
/**
* 回話的超時時間(毫秒)
*/
private Integer sessionTimeOut;
/**
* 連線的超時時間(毫秒)
*/
private Integer connectionTimeOut;
/**
* 名稱空間
**/
private String nameSpace;
/**
* zookeeper管理物件
*/
private CuratorFramework zkTools;
/**
* 獨享佇列節點
*/
private String monopolyQueueNode;
/**
* 連線狀態
*/
private String connectionState;
/**
* 會話ID
*/
private long sessionId;
/**
* Spring初始化方法
*/
public void afterPropertiesSet(){
this.connection();
this.addListener();
}
/**
* 連線
*/
public void connection(){
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, Integer.MAX_VALUE);
zkTools = CuratorFrameworkFactory
.builder()
.connectString(hosts)
.namespace(nameSpace)
.retryPolicy(retryPolicy)
.connectionTimeoutMs(connectionTimeOut == null ? 30000 : connectionTimeOut)
.sessionTimeoutMs(sessionTimeOut == null ? 30000 : sessionTimeOut)
.build();
zkTools.start();
}
/**
* 連線狀態監聽
*/
public void addListener(){
zkTools.getConnectionStateListenable().addListener(new ConnectionStateListener() {
public void stateChanged(CuratorFramework client, ConnectionState newState) {
if (newState.equals(ConnectionState.CONNECTED)) {
logger.info("連線");
connectionState = "CONNECTED";
try {
sessionId = zkTools.getZookeeperClient().getZooKeeper().getSessionId();
registerMonopolyQueue();
} catch (Exception e) {
logger.error("註冊獨佔佇列失敗");
}
}
if (newState.equals(ConnectionState.RECONNECTED)) {
logger.info("重新連線");
connectionState = "CONNECTED";
try {
if(sessionId != zkTools.getZookeeperClient().getZooKeeper().getSessionId()) {
registerMonopolyQueue();
}
} catch (Exception e) {
logger.error("註冊獨佔佇列失敗");
}
}
if (newState.equals(ConnectionState.LOST)) {
logger.info("丟失");
connectionState = "LOST";
}
if (newState.equals(ConnectionState.SUSPENDED)) {
logger.info("暫停");
connectionState = "SUSPENDED";
}
if (newState.equals(ConnectionState.READ_ONLY)) {
logger.info("只讀");
connectionState = "READ_ONLY";
}
}
});
}
/**
* 註冊獨佔佇列
*/
private void registerMonopolyQueue() throws Exception {
if(zkTools.checkExists().watched().forPath(Constant.MONOPOLY) == null){
zkTools.create()
.withMode(CreateMode.PERSISTENT)
.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
.forPath(Constant.MONOPOLY);
logger.info("建立獨享鎖佇列節點成功！");
}
if(monopolyQueueNode == null || (monopolyQueueNode != null && zkTools.checkExists().forPath(monopolyQueueNode)==null)) {
monopolyQueueNode = zkTools.create()
.withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
.forPath(Constant.MONOPOLY + Constant.SEPARATOR + Constant.QUEUE_NODE);
logger.info("成功加入獨享鎖佇列");
}
}
/**
* 獲得獨佔鎖的執行許可權
* @return 執行許可權標識
* @throws KeeperException
* @throws InterruptedException
*/
public boolean getMonopolyLock() throws Exception {
boolean flag = false;
if(connectionState != null && (connectionState.equals("CONNECTED") || connectionState.equals("RECONNECTED"))){
List<String> nodes = zkTools.getChildren().watched().forPath(Constant.MONOPOLY);
if(nodes.size() > 0){
Collections.sort(nodes);
//判斷當前應用是否在佇列的第一位
if((Constant.SEPARATOR + Constant.MONOPOLY + Constant.SEPARATOR + nodes.get(0)).equals(monopolyQueueNode)){
flag = true;
}
}
}
return flag;
}
/**
* 關閉連線
*/
public void close(){
if(zkTools != null){
zkTools.close();
zkTools = null;
}
}
public String getHosts() {
return hosts;
}
public void setHosts(String hosts) {
this.hosts = hosts;
}
public Integer getSessionTimeOut() {
return sessionTimeOut;
}
public void setSessionTimeOut(Integer sessionTimeOut) {
this.sessionTimeOut = sessionTimeOut;
}
public Integer getConnectionTimeOut() {
return connectionTimeOut;
}
public void setConnectionTimeOut(Integer connectionTimeOut) {
this.connectionTimeOut = connectionTimeOut;
}
public String getNameSpace() {
return nameSpace;
}
public void setNameSpace(String nameSpace) {
this.nameSpace = nameSpace;
}
}
package com.ab.scheduling.quartz.common;
import com.ab.scheduling.quartz.JobDetailProxyBean;
import com.ab.scheduling.quartz.CronTriggerBean;
import com.ab.scheduling.quartz.SimpleTriggerBean;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.simpl.SimpleThreadPool;
import java.util.Properties;
/**
* Quartz排程工具類
* Date: 14-5-15
* Time: 下午6:10
*/
public class SchedulerUtils {
protected static Logger logger = Logger.getLogger(SchedulerUtils.class);
/**
* 初始化StdSchedulerFactory
* @return StdSchedulerFactory
*/
public static StdSchedulerFactory initStdSchedulerFactory() {
StdSchedulerFactory schedulerFactory = null;
try{
schedulerFactory = (StdSchedulerFactory) Class.forName(StdSchedulerFactory.class.getName()).newInstance();
Properties mergedProps = new Properties();
// 設定Quartz執行緒池設定
mergedProps.setProperty(StdSchedulerFactory.PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName());
mergedProps.setProperty("org.quartz.threadPool.threadCount", Integer.toString(10));
schedulerFactory.initialize(mergedProps);
} catch (Exception e){
logger.error("初始化StdSchedulerFactory失敗");
logger.error(e);
}
return schedulerFactory;
}
/**
* 裝配任務
* @param jobDetail 任務代理類
* @return JobDetail
*/
public static JobDetail assemblyJobDetail(JobDetailProxyBean jobDetail){
JobBuilder jobBuilder = JobBuilder.newJob(jobDetail.getClass());
//設定JobDetail身份標識與所屬組
String key = jobDetail.getKey();
if(StringUtils.isNotBlank(key)){
jobBuilder = jobBuilder.withIdentity(key, jobDetail.getGroup());
}else{
jobBuilder = jobBuilder.withIdentity(IdentityUtils.generatorUUID("JOB"), jobDetail.getGroup());
}
//設定任務描述
if(StringUtils.isNotBlank(jobDetail.getDescription())){
jobBuilder = jobBuilder.withDescription(jobDetail.getDescription());
}
//設定JobDetail資料引數
JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put("targetObject",jobDetail.getTargetObject()); //目標物件
jobDataMap.put("targetMethod",jobDetail.getTargetMethod()); //目標方法
jobDataMap.put("mode", jobDetail.getMode()); //執行模式
jobBuilder = jobBuilder.usingJobData(jobDataMap);
return jobBuilder.build();
}
/**
* 裝配表示式觸發器
* @param cronTriggerBean 表示式觸發器
* @return 表示式觸發器
*/
public static CronTrigger assemblyCronTrigger(CronTriggerBean cronTriggerBean){
TriggerBuilder triggerBuilder = TriggerBuilder.newTrigger();
//設定觸發器身份標識與所屬組
String key = cronTriggerBean.getKey();
if(StringUtils.isNotBlank(key)){
triggerBuilder = triggerBuilder.withIdentity(key, cronTriggerBean.getGroup());
}else{
triggerBuilder = triggerBuilder.withIdentity(IdentityUtils.generatorUUID("CronTrigger"), cronTriggerBean.getGroup());
}
//設定描述
if(StringUtils.isNotBlank(cronTriggerBean.getDescription())){
triggerBuilder = triggerBuilder.withDescription(cronTriggerBean.getDescription());
}
//設定啟動時間
if(StringUtils.isNotBlank(cronTriggerBean.getStartTime())){
triggerBuilder = triggerBuilder.startAt(DateUtils.StringToDate(cronTriggerBean.getStartTime(), "yyyy-MM-dd HH:mm:ss"));
}else{
triggerBuilder = triggerBuilder.startNow(); //當啟動時間為空預設立即啟動排程器
}
//設定結束時間
if(StringUtils.isNotBlank(cronTriggerBean.getEndTime())){
triggerBuilder = triggerBuilder.endAt(DateUtils.StringToDate(cronTriggerBean.getEndTime(), "yyyy-MM-dd HH:mm:ss"));
}
//設定優先順序
if(cronTriggerBean.getPriority() != null){
triggerBuilder = triggerBuilder.withPriority(cronTriggerBean.getPriority());
}
//設定Cron表示式(不允許為空)與集火指令
CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(cronTriggerBean.getCronExpression());
if(cronTriggerBean.getMisfireInstruction() != null){
if(cronTriggerBean.getMisfireInstruction() == Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY) {
cronScheduleBuilder = cronScheduleBuilder.withMisfireHandlingInstructionIgnoreMisfires();
}
if(cronTriggerBean.getMisfireInstruction() == CronTrigger.MISFIRE_INSTRUCTION_FIRE_ONCE_NOW) {
cronScheduleBuilder = cronScheduleBuilder.withMisfireHandlingInstructionFireAndProceed();
}
if(cronTriggerBean.getMisfireInstruction() == CronTrigger.MISFIRE_INSTRUCTION_DO_NOTHING) {
cronScheduleBuilder = cronScheduleBuilder.withMisfireHandlingInstructionDoNothing();
}
}
triggerBuilder = triggerBuilder.withSchedule(cronScheduleBuilder);
return (CronTrigger)triggerBuilder.build();
}
/**
* 裝配簡單觸發器
* @param simpleTriggerBean 簡單觸發器
* @return 簡單觸發器
*/
public static SimpleTrigger assemblySimpleTrigger(SimpleTriggerBean simpleTriggerBean){
TriggerBuilder triggerBuilder = TriggerBuilder.newTrigger();
//設定觸發器身份標識與所屬組
String key = simpleTriggerBean.getKey();
if(StringUtils.isNotBlank(key)){
triggerBuilder = triggerBuilder.withIdentity(key, simpleTriggerBean.getGroup());
}else{
triggerBuilder = triggerBuilder.withIdentity(IdentityUtils.generatorUUID("SimpleTrigger"), simpleTriggerBean.getGroup());
}
//設定描述
if(StringUtils.isNotBlank(simpleTriggerBean.getDescription())){
triggerBuilder = triggerBuilder.withDescription(simpleTriggerBean.getDescription());
}
//設定啟動時間
if(StringUtils.isNotBlank(simpleTriggerBean.getStartTime())){
triggerBuilder = triggerBuilder.startAt(DateUtils.StringToDate(simpleTriggerBean.getStartTime(), "yyyy-MM-dd HH:mm:ss"));
}else{
triggerBuilder = triggerBuilder.startNow(); //當啟動時間為空預設立即啟動排程器
}
//設定結束時間
if(StringUtils.isNotBlank(simpleTriggerBean.getEndTime())){
triggerBuilder = triggerBuilder.endAt(DateUtils.StringToDate(simpleTriggerBean.getEndTime(), "yyyy-MM-dd HH:mm:ss"));
}
//設定優先順序
if(simpleTriggerBean.getPriority() != null){
triggerBuilder = triggerBuilder.withPriority(simpleTriggerBean.getPriority());
}
//設定簡單觸發器 時間間隔(不允許為空)、執行次數(預設為-1)與集火指令
SimpleScheduleBuilder simpleScheduleBuilder = SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(20).withRepeatCount(-1);
simpleScheduleBuilder = simpleScheduleBuilder.withIntervalInSeconds(simpleTriggerBean.getInterval());
if(simpleTriggerBean.getRepeatCount() != null){
simpleScheduleBuilder = simpleScheduleBuilder.withRepeatCount(simpleTriggerBean.getRepeatCount());
}else{
simpleScheduleBuilder = simpleScheduleBuilder.withRepeatCount(-1);
}
if(simpleTriggerBean.getMisfireInstruction() != null){
if(simpleTriggerBean.getMisfireInstruction() == Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY) {
simpleScheduleBuilder = simpleScheduleBuilder.withMisfireHandlingInstructionIgnoreMisfires();
}
if(simpleTriggerBean.getMisfireInstruction() == SimpleTrigger.MISFIRE_INSTRUCTION_FIRE_NOW) {
simpleScheduleBuilder = simpleScheduleBuilder.withMisfireHandlingInstructionFireNow();
}
if(simpleTriggerBean.getMisfireInstruction() == SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_EXISTING_COUNT) {
simpleScheduleBuilder = simpleScheduleBuilder.withMisfireHandlingInstructionNextWithExistingCount();
}
if(simpleTriggerBean.getMisfireInstruction() == SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT) {
simpleScheduleBuilder = simpleScheduleBuilder.withMisfireHandlingInstructionNextWithRemainingCount();
}
if(simpleTriggerBean.getMisfireInstruction() == SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT) {
simpleScheduleBuilder = simpleScheduleBuilder.withMisfireHandlingInstructionNowWithExistingCount();
}
if(simpleTriggerBean.getMisfireInstruction() == SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_REMAINING_REPEAT_COUNT) {
simpleScheduleBuilder = simpleScheduleBuilder.withMisfireHandlingInstructionNowWithRemainingCount();
}
}
triggerBuilder = triggerBuilder.withSchedule(simpleScheduleBuilder);
return (SimpleTrigger)triggerBuilder.build();
}
}
spring配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"
default-autowire="byName">
<!--定時任務實現類-->
<bean id="test1" class="com.jd.scheduling.quartz.test.Test1"/>
<!--任務代理-->
<bean id="jobDetail1" class="com.ab.scheduling.quartz.JobDetailProxyBean">
<property name="targetObject" ref="test1"/>
<property name="targetMethod" value="test"/>
</bean>
<!--觸發器-->
<bean id="cronTrigger" class="com.ab.scheduling.quartz.CronTriggerBean">
<property name="jobDetail" ref="jobDetail1"/>
<property name="cronExpression" value="0/10 * * * * ?"/>
</bean>
<!--zk配置-->
<bean id="zooKeeperFactory" class="com.ab.scheduling.quartz.ZookeeperFactory">
<property name="hosts" value="127.0.0.1:2181"/>
<property name="sessionTimeOut" value="15000"/>
<property name="nameSpace" value="zk-scheduling"/>
</bean>
<!--排程工廠-->
<bean id="schdulerFactory" autowire="no" class="com.ab.scheduling.quartz.SchedulerFactoryBean">
<property name="triggers">
<list>
<ref bean="cronTrigger" />
</list>
</property>
<property name="zooKeeperFactory" ref="zooKeeperFactory"/>
</bean>
</beans>