定时任务分布式锁的简单实现
在集群环境下,若每一台机器都运行一个定时任务,会导致生产数据一致性问题,所以必须要实现一个锁。保证当时任务在同一时间段只能在一台机器上面运行。有的同学应该已经想到分布式锁了,例如用redis或者zookeeper来实现分布式锁。下面我介绍一种最简单的实现定时任务互斥执行的机制,那就是使用数据库update互斥的原理。运行环境:springMvc+quartz+m
·
在集群环境下,若每一台机器都运行一个定时任务,会导致生产数据一致性问题,所以必须要实现一个锁。保证当时任务在同一时间段只能在一台机器上面运行。
有的同学应该已经想到分布式锁了,例如用redis或者zookeeper来实现分布式锁。
下面我介绍一种最简单的实现定时任务互斥执行的机制,那就是使用数据库乐观锁的原理。
运行环境:springMvc+quartz+mybatis
package com.test.job;
import com.test.common.constants.Constants;
import com.test.common.util.BlankUtil;
import com.test.common.util.DateUtil;
import com.test.common.dao.BaseJobConfigMapper;
import com.test.common.dao.BaseJobConfigRecordMapper;
import com.test.model.BaseJobConfig;
import com.test.model.BaseJobConfigRecord;
import com.test.utils.SpringContextUtil;
import org.apache.log4j.Logger;
import org.quartz.CronTrigger;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.Date;
@Component
public abstract class BaseJob extends QuartzJobBean {
private static final Logger logger = Logger.getLogger(BaseJob.class);
@Autowired
private static BaseJobConfigMapper baseJobConfigMapper;
@Autowired
private static BaseJobConfigRecordMapper baseJobConfigRecordMapper;
@Autowired
private static DataSourceTransactionManager transactionManager;
public static String IP_STRING = null;
static {
if (baseJobConfigMapper == null) {
baseJobConfigMapper = (BaseJobConfigMapper) SpringContextUtil.getBean("baseJobConfigMapper");
}
if (baseJobConfigRecordMapper == null) {
baseJobConfigRecordMapper = (BaseJobConfigRecordMapper) SpringContextUtil.getBean("baseJobConfigRecordMapper");
}
if (transactionManager == null) {
transactionManager = (DataSourceTransactionManager) SpringContextUtil.getBean("transactionManager");
}
try {
InetAddress ip = InetAddress.getLocalHost();
if (!BlankUtil.isBlank(ip)) {
IP_STRING = ip.getHostAddress();
logger.info("本机地址" + IP_STRING);
}
} catch (UnknownHostException e) {
logger.error(e.getMessage(), e);
}
}
/**
* Job名称
*/
public String JOB_NAME = getJobName();
/**
* 重置时间--分钟
*/
public int JOB_RESET_TIME = resetJobTime();
/**
* 要调度的具体任务
*/
@Override
@Transactional
protected void executeInternal(JobExecutionContext context) {
if (!Constants.IS_DEV_MODE) {
//1、先判断JOB_NAME是否不为空,为空则结束
if (!BlankUtil.isBlank(JOB_NAME)) {
CronTrigger cTrigger = (CronTrigger) context.getTrigger();
SimpleDateFormat format = new SimpleDateFormat(DateUtil.DEFAULT_DATE_TIME);
Date triggerTime = cTrigger.getPreviousFireTime();
String currentTime = format.format(triggerTime);
BaseJobConfig selectBaseJobConfig = new BaseJobConfig();
selectBaseJobConfig.setKeyName(JOB_NAME);
long numLong = baseJobConfigMapper.selectCount(selectBaseJobConfig);
//3、若是没有,则插入,状态为0(待执行)
if (numLong <= 0) {
BaseJobConfig baseJobConfig = new BaseJobConfig();
baseJobConfig.setKeyName(JOB_NAME);
//初始化触发时间是上一个小时,避免为当前时间时,被误认为已经跑过
baseJobConfig.setSchedulePreTime(DateUtil.getDateByDifferHours(triggerTime, -1));
baseJobConfig.setCreateTime(new Date());
baseJobConfig.setState(0);
baseJobConfigMapper.insert(baseJobConfig);
}
int numInt = 0;
//30分钟未执行完,则重置状态
numInt = baseJobConfigMapper.updateBaseJobConfigStatusByTimeOut(JOB_NAME, JOB_RESET_TIME);
logger.info("[baseJob.doJob]job.key.resertjobName=" + JOB_NAME + ",num=" + numInt);
//有状态为0的待执行数据,则将状态修改为1(执行中)
numInt = baseJobConfigMapper.updateBaseJobConfigStatusAtStartTime(JOB_NAME, currentTime);
if (numInt > 0) {
DefaultTransactionDefinition def = new DefaultTransactionDefinition();
def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
TransactionStatus status = transactionManager.getTransaction(def);
Date startTime = new Date();
//执行业务逻辑
try {
logger.info("[baseJob.doJob]执行开始jobName=" + JOB_NAME);
jobExecute();
transactionManager.commit(status);
} catch (Exception e) {
logger.error(e.getMessage(), e);
transactionManager.rollback(status);
} finally {
//5、业务逻辑执行完后,将状态修改为0
numInt = baseJobConfigMapper.updateBaseJobConfigStatusAtEndTime(JOB_NAME, currentTime);
if (numInt <= 0) {
logger.error("[baseJob.doJob]active update error jobName=" + JOB_NAME);
}
Date endTime = new Date();
//记录定时任务运行情况
recordBaseJob(JOB_NAME, triggerTime, startTime, endTime, (endTime.getTime() - startTime.getTime()), IP_STRING);
logger.info("[baseJob.doJob]执行结束jobName=" + JOB_NAME + "------" + (endTime.getTime() - startTime.getTime()));
}
}
} else {
logger.error("[baseJob.doJob]JOB_NAME is null");
}
}
}
public abstract void jobExecute() throws Exception;
public abstract String getJobName();
public int resetJobTime() {
return 30;
}
/**
* 功能描述:
* 记录任务运行情况
*
*/
private void recordBaseJob(String keyName, Date triggerTime, Date startTime, Date endTime, Long costTime, String ip) {
try {
BaseJobConfigRecord baseJobConfigRecord = new BaseJobConfigRecord();
baseJobConfigRecord.setKeyName(keyName);
baseJobConfigRecord.setTriggerTime(triggerTime);
baseJobConfigRecord.setStartTime(startTime);
baseJobConfigRecord.setEndTime(endTime);
baseJobConfigRecord.setCostTime(costTime);
baseJobConfigRecord.setIp(ip);
baseJobConfigRecord.setCreateTime(new Date());
baseJobConfigRecordMapper.insert(baseJobConfigRecord);
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
package com.test.job;
import com.test.job.BaseJob;
import org.apache.log4j.Logger;
/**
* 功能描述:
* 测试定时任务
*/
public class TestJob extends BaseJob {
private static final Logger logger = Logger.getLogger(TestJob.class);
@Override
public void jobExecute() throws Exception {
System.out.println("测试定时任务");
}
@Override
public String getJobName() {
return "TestJob";
}
}
<!-- 定时任务注解配置 -->
<task:annotation-driven scheduler="scheduler" mode="proxy"/>
<task:scheduler id="scheduler" pool-size="10"/>
spring-job.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"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="trigger" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="triggers">
<list>
<ref bean="testJobCronTriggerBean"/>
</list>
</property>
</bean>
<!--测试任务 start-->
<bean id="testJobDetail" class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
<property name="jobClass" value="com.test.job.TestJob"/>
<property name="durability" value="true"/>
</bean>
<bean id="testJobCronTriggerBean"
class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
<property name="jobDetail" ref="testJobDetail"/>
<property name="cronExpression" value="0 0/1 * * * ?"/>
</bean>
<!--测试任务 end-->
</beans>
mybatis相关代码如下:
package com.test.common.dao;
import com.test.model.BaseJobConfig;
import org.apache.ibatis.annotations.Param;
import tk.mybatis.mapper.common.Mapper;
public interface BaseJobConfigMapper extends Mapper<BaseJobConfig> {
int updateBaseJobConfigStatusByTimeOut(@Param("keyName") String keyName, @Param("resetTime") Integer resetTime);
int updateBaseJobConfigStatusAtStartTime(@Param("keyName") String keyName, @Param("schedulePreTime") String schedulePreTime);
int updateBaseJobConfigStatusAtEndTime(@Param("keyName") String keyName, @Param("schedulePreTime") String schedulePreTime);
}
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.test.common.dao.BaseJobConfigMapper">
<resultMap id="BaseResultMap" type="com.test.model.BaseJobConfig">
<!--
WARNING - @mbg.generated
-->
<id column="key_name" jdbcType="VARCHAR" property="keyName" />
<result column="key_value" jdbcType="VARCHAR" property="keyValue" />
<result column="schedule_pre_time" jdbcType="TIMESTAMP" property="schedulePreTime" />
<result column="actual_pre_time" jdbcType="TIMESTAMP" property="actualPreTime" />
<result column="state" jdbcType="TINYINT" property="state" />
<result column="create_time" jdbcType="TIMESTAMP" property="createTime" />
</resultMap>
<!-- 运行超时,重置状态 -->
<update id="updateBaseJobConfigStatusByTimeOut">
<![CDATA[
update basejob_config set state = 0 where key_name = #{keyName} and state = 1 and actual_pre_time < DATE_SUB(now(), INTERVAL #{resetTime} MINUTE)
]]>
</update>
<!-- 定时任务运行开始更新互斥状态 -->
<update id="updateBaseJobConfigStatusAtStartTime">
<![CDATA[
update basejob_config set state = 1,actual_pre_time = now(),schedule_pre_time = #{schedulePreTime} where key_name = #{keyName} and state = 0 and schedule_pre_time != #{schedulePreTime}
]]>
</update>
<!-- 定时任务运行结束后更新互斥状态 -->
<update id="updateBaseJobConfigStatusAtEndTime">
<![CDATA[
update basejob_config set state = 0 where key_name = #{keyName} and state = 1 and schedule_pre_time = #{schedulePreTime}
]]>
</update>
</mapper>
数据库表设计:
CREATE TABLE `basejob_config` (
`key_name` varchar(255) NOT NULL DEFAULT '' COMMENT '参数code',
`key_value` varchar(255) DEFAULT NULL COMMENT '参数值',
`schedule_pre_time` datetime DEFAULT NULL COMMENT '上一次计划运行时间',
`actual_pre_time` datetime DEFAULT NULL COMMENT '上一次实际执行时间',
`state` tinyint(4) NOT NULL DEFAULT '0' COMMENT '状态--1代表正在执行0代表等待执行',
`create_time` datetime DEFAULT NULL COMMENT '创建时间',
PRIMARY KEY (`key_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='baseJob配置表';
CREATE TABLE `basejob_config_record` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键',
`key_name` varchar(255) DEFAULT NULL COMMENT '定时任务名称',
`trigger_time` datetime DEFAULT NULL COMMENT '定时任务计划触发时间',
`start_time` datetime DEFAULT NULL COMMENT '定时任务开始时间',
`end_time` datetime DEFAULT NULL COMMENT '定时任务结束时间',
`cost_time` bigint(20) DEFAULT NULL COMMENT '耗时',
`ip` varchar(50) DEFAULT NULL COMMENT '运行服务器IP',
`create_time` datetime DEFAULT NULL COMMENT '创建时间',
PRIMARY KEY (`id`),
KEY `idx_basejob_config_record_key_name` (`key_name`) USING BTREE,
KEY `idx_basejob_config_record_trigger_time` (`trigger_time`) USING BTREE,
KEY `idx_basejob_config_record_start_time` (`start_time`) USING BTREE,
KEY `idx_basejob_config_record_end_time` (`end_time`) USING BTREE,
KEY `idx_basejob_config_record_create_time` (`create_time`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='定时任务运行记录表';
原理就是利用数据库乐观锁对数据库行记录进行update,若能update成功,则证明服务器抢到一个锁,则执行定时任务,update不成功的服务器,则直接退出,这样一个简单的定时任务分布式锁就实现了。
更多推荐
已为社区贡献1条内容
所有评论(0)