java 业务幂等解决方案
-
现在的应用基本都是分布式部署,所有机器的代码都是一样的,前面介绍的 Java 和 Spring 自带的解决方案,都是进程级别的,每台机器在同一时间点都会执行定时任务。这样会导致需要业务幂等的定时任务业务有问题,比如每月定时给用户推送消息,就会推送多次。 -
于是,很多应用很自然的就想到了使用分布式锁的解决方案。即每次定时任务执行之前,先去抢锁,抢到锁的执行任务,抢不到锁的不执行。怎么抢锁,又是五花八门,比如使用 DB、zookeeper、redis。
使用 DB 或者 Zookeeper 抢锁
-
使用 DB 或者 Zookeeper 抢锁的架构差不多,原理如下: -
 -
定时时间到了,在回调方法里,先去抢锁。 ? -
抢到锁,则继续执行方法,没抢到锁直接返回。 -
执行完方法后,释放锁。 -
示例代码如下: -
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
?
@Component
@EnableScheduling
public class MyTask {
? ?/**
? ? * 每分钟的第30秒跑一次
? ? */
? ?@Scheduled(cron = "30 * * * * ?")
? ?public void task1() throws Exception {
? ? ? ?String lockName = "task1";
? ? ? ?if (tryLock(lockName)) {
? ? ? ? ? ?System.out.println("hello cron");
? ? ? ? ? ?releaseLock(lockName);
? ? ? } else {
? ? ? ? ? ?return;
? ? ? }
? }
?
? ?private boolean tryLock(String lockName) {
? ? ? ?//TODO
? ? ? ?return true;
? }
?
? ?private void releaseLock(String lockName) {
? ? ? ?//TODO
? }
} -
当前的这个设计,仔细一点的同学可以发现,其实还是有可能导致任务重复执行的。比如任务执行的非常快,A 这台机器抢到锁,执行完任务后很快就释放锁了。B 这台机器后抢锁,还是会抢到锁,再执行一遍任务。
使用 redis 抢锁
-
使用 redis 抢锁,其实架构上和 DB/zookeeper 差不多,不过 redis 抢锁支持过期时间,不用主动去释放锁,并且可以充分利用这个过期时间,解决任务执行过快释放锁导致任务重复执行的问题,架构如下: -
 -
示例代码如下: -
@Component
@EnableScheduling
public class MyTask {
? ?/**
? ? * 每分钟的第30秒跑一次
? ? */
? ?@Scheduled(cron = "30 * * * * ?")
? ?public void task1() throws InterruptedException {
? ? ? ?String lockName = "task1";
? ? ? ?if (tryLock(lockName, 30)) {
? ? ? ? ? ?System.out.println("hello cron");
? ? ? ? ? ?releaseLock(lockName);
? ? ? } else {
? ? ? ? ? ?return;
? ? ? }
? }
?
? ?private boolean tryLock(String lockName, long expiredTime) {
? ? ? ?//TODO
? ? ? ?return true;
? }
?
? ?private void releaseLock(String lockName) {
? ? ? ?//TODO
? }
} -
看到这里,可能又会有同学有问题,加一个过期时间是不是还是不够严谨,还是有可能任务重复执行? -
——的确是的,如果有一台机器突然长时间的 fullgc,或者之前的任务还没处理完(Spring Task 和 ScheduledExecutorService 本质还是通过线程池处理任务),还是有可能隔了 30 秒再去调度任务的。
使用 Quartz
-
Quartz [ 1] 是一套轻量级的任务调度框架,只需要定义了 Job(任务),Trigger(触发器)和 Scheduler(调度器),即可实现一个定时调度能力。支持基于数据库的集群模式,可以做到任务幂等执行。 -
 -
Quartz 支持任务幂等执行,其实理论上还是抢 DB 锁,我们看下 quartz 的表结构: -
 -
其中,QRTZ_LOCKS 就是 Quartz 集群实现同步机制的行锁表,其表结构如下: -
--QRTZ_LOCKS表结构
CREATE TABLE `QRTZ_LOCKS` (
`LOCK_NAME` varchar(40) NOT NULL,
PRIMARY KEY (`LOCK_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
?
--QRTZ_LOCKS记录
+-----------------+
| LOCK_NAME ? ? ? |
+-----------------+
| CALENDAR_ACCESS |
| JOB_ACCESS ? ? |
| MISFIRE_ACCESS |
| STATE_ACCESS ? |
| TRIGGER_ACCESS |
+-----------------+ -
可以看出 QRTZ_LOCKS 中有 5 条记录,代表 5 把锁,分别用于实现多个 Quartz Node 对 Job、Trigger、Calendar 访问的同步控制。
ElasticJob
-
ElasticJob [ 2] 是一款基于 Quartz 开发,依赖 Zookeeper 作为注册中心、轻量级、无中心化的分布式任务调度框架,目前已经通过 Apache 开源。 -
ElasticJob 相对于 Quartz 来说,从功能上最大的区别就是支持分片,可以将一个任务分片参数分发给不同的机器执行。架构上最大的区别就是使用 Zookeeper 作为注册中心,不同的任务分配给不同的节点调度,不需要抢锁触发,性能上比 Quartz 上强大很多,架构图如下: -
 -
开发上也比较简单,和 springboot 结合比较好,可以在配置文件定义任务如下: -
elasticjob:
regCenter:
? serverLists: localhost:2181
? namespace: elasticjob-lite-springboot
jobs:
? simpleJob:
? ? elasticJobClass: org.apache.shardingsphere.elasticjob.lite.example.job.SpringBootSimpleJob
? ? cron: 0/5 * * * * ?
? ? timeZone: GMT+08:00
? ? shardingTotalCount: 3
? ? shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou
? scriptJob:
? ? elasticJobType: SCRIPT
? ? cron: 0/10 * * * * ?
? ? shardingTotalCount: 3
? ? props:
? ? ? script.command.line: "echo SCRIPT Job: "
? manualScriptJob:
? ? elasticJobType: SCRIPT
? ? jobBootstrapBeanName: manualScriptJobBean
? ? shardingTotalCount: 9
? ? props:
? ? ? script.command.line: "echo Manual SCRIPT Job: " -
实现任务接口如下: -
@Component
public class SpringBootShardingJob implements SimpleJob {
?
? ?@Override
? ?public void execute(ShardingContext context) {
? ? ? ?System.out.println("分片总数="+context.getShardingTotalCount() + ", 分片号="+context.getShardingItem()
? ? ? ? ? ?+ ", 分片参数="+context.getShardingParameter());
? }
} -
运行结果如下: -
分片总数=3, 分片号=0, 分片参数=Beijing
分片总数=3, 分片号=1, 分片参数=Shanghai
分片总数=3, 分片号=2, 分片参数=Guangzhou -
同时,ElasticJob 还提供了一个简单的 UI,可以查看任务的列表,同时支持修改、触发、停止、生效、失效操作。 -
 -
遗憾的是,ElasticJob 暂不支持动态创建任务。
XXL-JOB
-
XXL-JOB [ 3] 是一个开箱即用的轻量级分布式任务调度系统,其核心设计目标是开发迅速、学习简单、轻量级、易扩展,在开源社区广泛流行。 -
XXL-JOB 是 Master-Slave 架构,Master 负责任务的调度,Slave 负责任务的执行,架构图如下: -
 -
XXL-JOB 接入也很方便,不同于 ElasticJob 定义任务实现类,是通过@XxlJob 注解定义 JobHandler。 -
@Component
public class SampleXxlJob {
? ?private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);
? ?/**
? ? * 1、简单任务示例(Bean模式)
? ? */
? ?@XxlJob("demoJobHandler")
? ?public ReturnT<String> demoJobHandler(String param) throws Exception {
? ? ? ?XxlJobLogger.log("XXL-JOB, Hello World.");
?
?
? ? ? ?for (int i = 0; i < 5; i++) {
? ? ? ? ? ?XxlJobLogger.log("beat at:" + i);
? ? ? ? ? ?TimeUnit.SECONDS.sleep(2);
? ? ? }
? ? ? ?return ReturnT.SUCCESS;
? }
?
? ?/**
? ? * 2、分片广播任务
? ? */
? ?@XxlJob("shardingJobHandler")
? ?public ReturnT<String> shardingJobHandler(String param) throws Exception {
?
? ? ? ?// 分片参数
? ? ? ?ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
? ? ? ?XxlJobLogger.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());
?
? ? ? ?// 业务逻辑
? ? ? ?for (int i = 0; i < shardingVO.getTotal(); i++) {
? ? ? ? ? ?if (i == shardingVO.getIndex()) {
? ? ? ? ? ? ? ?XxlJobLogger.log("第 {} 片, 命中分片开始处理", i);
? ? ? ? ? } else {
? ? ? ? ? ? ? ?XxlJobLogger.log("第 {} 片, 忽略", i);
? ? ? ? ? }
? ? ? }
? ? ? ?return ReturnT.SUCCESS;
? }
}
? -
XXL-JOB 相较于 ElasticJob,最大的特点就是功能比较丰富,可运维能力比较强,不但支持控制台动态创建任务,还有调度日志、运行报表等功能。 -
 ? -
 -
XXL-JOB 的历史记录、运行报表和调度日志,都是基于数据库实现的: -
 -
由此可以看出,XXL-JOB 所有功能都依赖数据库,且调度中心不支持分布式架构,在任务量和调度量比较大的情况下,会有性能瓶颈。不过如果对任务量级、高可用、监控报警、可视化等没有过高要求的话,XXL-JOB 基本可以满足定时任务的需求。
|