• java 业务幂等解决方案

    java 业务幂等解决方案

    • 现在的应用基本都是分布式部署，所有机器的代码都是一样的，前面介绍的 Java 和 Spring 自带的解决方案，都是进程级别的，每台机器在同一时间点都会执行定时任务。这样会导致需要业务幂等的定时任务业务有问题，比如每月定时给用户推送消息，就会推送多次。

    • 于是，很多应用很自然的就想到了使用分布式锁的解决方案。即每次定时任务执行之前，先去抢锁，抢到锁的执行任务，抢不到锁的不执行。怎么抢锁，又是五花八门，比如使用 DB、zookeeper、redis。

    使用 DB 或者 Zookeeper 抢锁

    • 使用 DB 或者 Zookeeper 抢锁的架构差不多，原理如下：

    • 定时时间到了，在回调方法里，先去抢锁。

       

    • 抢到锁，则继续执行方法，没抢到锁直接返回。

    • 执行完方法后，释放锁。

    • 示例代码如下：

      1. import org.springframework.scheduling.annotation.EnableScheduling;
      2. import org.springframework.scheduling.annotation.Scheduled;
      3. import org.springframework.stereotype.Component;
      4. ​
      5. @Component
      6. @EnableScheduling
      7. public class MyTask {
      8.     /**
      9.      * 每分钟的第30秒跑一次
      10.      */
      11.     @Scheduled(cron = "30 * * * * ?")
      12.     public void task1() throws Exception {
      13.         String lockName = "task1";
      14.         if (tryLock(lockName)) {
      15.             System.out.println("hello cron");
      16.             releaseLock(lockName);
      17.         } else {
      18.             return;
      19.         }
      20.     }
      21. ​
      22.     private boolean tryLock(String lockName) {
      23.         //TODO
      24.         return true;
      25.     }
      26. ​
      27.     private void releaseLock(String lockName) {
      28.         //TODO
      29.     }
      30. }

    • 当前的这个设计，仔细一点的同学可以发现，其实还是有可能导致任务重复执行的。比如任务执行的非常快，A 这台机器抢到锁，执行完任务后很快就释放锁了。B 这台机器后抢锁，还是会抢到锁，再执行一遍任务。

    使用 redis 抢锁

    • 使用 redis 抢锁，其实架构上和 DB/zookeeper 差不多，不过 redis 抢锁支持过期时间，不用主动去释放锁，并且可以充分利用这个过期时间，解决任务执行过快释放锁导致任务重复执行的问题，架构如下：

    • 示例代码如下：

      1. @Component
      2. @EnableScheduling
      3. public class MyTask {
      4.     /**
      5.      * 每分钟的第30秒跑一次
      6.      */
      7.     @Scheduled(cron = "30 * * * * ?")
      8.     public void task1() throws InterruptedException {
      9.         String lockName = "task1";
      10.         if (tryLock(lockName, 30)) {
      11.             System.out.println("hello cron");
      12.             releaseLock(lockName);
      13.         } else {
      14.             return;
      15.         }
      16.     }
      17. ​
      18.     private boolean tryLock(String lockName, long expiredTime) {
      19.         //TODO
      20.         return true;
      21.     }
      22. ​
      23.     private void releaseLock(String lockName) {
      24.         //TODO
      25.     }
      26. }

    • 看到这里，可能又会有同学有问题，加一个过期时间是不是还是不够严谨，还是有可能任务重复执行？

    • ——的确是的，如果有一台机器突然长时间的 fullgc，或者之前的任务还没处理完（Spring Task 和 ScheduledExecutorService 本质还是通过线程池处理任务），还是有可能隔了 30 秒再去调度任务的。

    使用 Quartz

    • Quartz [ 1] 是一套轻量级的任务调度框架，只需要定义了 Job（任务），Trigger（触发器）和 Scheduler（调度器），即可实现一个定时调度能力。支持基于数据库的集群模式，可以做到任务幂等执行。

    • Quartz 支持任务幂等执行，其实理论上还是抢 DB 锁，我们看下 quartz 的表结构：

    • 其中，QRTZ_LOCKS 就是 Quartz 集群实现同步机制的行锁表，其表结构如下：

      1. --QRTZ_LOCKS表结构
      2. CREATE TABLE `QRTZ_LOCKS` (
      3.   `LOCK_NAME` varchar(40) NOT NULL,
      4.   PRIMARY KEY (`LOCK_NAME`)
      5. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
      6. ​
      7. --QRTZ_LOCKS记录
      8. +-----------------+ 
      9. | LOCK_NAME       |
      10. +-----------------+ 
      11. | CALENDAR_ACCESS |
      12. | JOB_ACCESS      |
      13. | MISFIRE_ACCESS  |
      14. | STATE_ACCESS    |
      15. | TRIGGER_ACCESS  |
      16. +-----------------+

    • 可以看出 QRTZ_LOCKS 中有 5 条记录，代表 5 把锁，分别用于实现多个 Quartz Node 对 Job、Trigger、Calendar 访问的同步控制。

    ElasticJob

    • ElasticJob [ 2] 是一款基于 Quartz 开发，依赖 Zookeeper 作为注册中心、轻量级、无中心化的分布式任务调度框架，目前已经通过 Apache 开源。

    • ElasticJob 相对于 Quartz 来说，从功能上最大的区别就是支持分片，可以将一个任务分片参数分发给不同的机器执行。架构上最大的区别就是使用 Zookeeper 作为注册中心，不同的任务分配给不同的节点调度，不需要抢锁触发，性能上比 Quartz 上强大很多，架构图如下：

    • 开发上也比较简单，和 springboot 结合比较好，可以在配置文件定义任务如下：

      1. elasticjob:
      2.   regCenter:
      3.     serverLists: localhost:2181
      4.     namespace: elasticjob-lite-springboot
      5.   jobs:
      6.     simpleJob:
      7.       elasticJobClass: org.apache.shardingsphere.elasticjob.lite.example.job.SpringBootSimpleJob
      8.       cron: 0/5 * * * * ?
      9.       timeZone: GMT+08:00
      10.       shardingTotalCount: 3
      11.       shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou
      12.     scriptJob:
      13.       elasticJobType: SCRIPT
      14.       cron: 0/10 * * * * ?
      15.       shardingTotalCount: 3
      16.       props:
      17.         script.command.line: "echo SCRIPT Job: "
      18.     manualScriptJob:
      19.       elasticJobType: SCRIPT
      20.       jobBootstrapBeanName: manualScriptJobBean
      21.       shardingTotalCount: 9
      22.       props:
      23.         script.command.line: "echo Manual SCRIPT Job: "

    • 实现任务接口如下：

      1. @Component
      2. public class SpringBootShardingJob implements SimpleJob {
      3. ​
      4.     @Override
      5.     public void execute(ShardingContext context) {
      6.         System.out.println("分片总数="+context.getShardingTotalCount() + ", 分片号="+context.getShardingItem()
      7.             + ", 分片参数="+context.getShardingParameter());
      8.     }
      9. }

    • 运行结果如下：

      1. 分片总数=3, 分片号=0, 分片参数=Beijing
      2. 分片总数=3, 分片号=1, 分片参数=Shanghai
      3. 分片总数=3, 分片号=2, 分片参数=Guangzhou

    • 同时，ElasticJob 还提供了一个简单的 UI，可以查看任务的列表，同时支持修改、触发、停止、生效、失效操作。

    • 遗憾的是，ElasticJob 暂不支持动态创建任务。

    XXL-JOB

    • XXL-JOB [ 3] 是一个开箱即用的轻量级分布式任务调度系统，其核心设计目标是开发迅速、学习简单、轻量级、易扩展，在开源社区广泛流行。

    • XXL-JOB 是 Master-Slave 架构，Master 负责任务的调度，Slave 负责任务的执行，架构图如下：

    • XXL-JOB 接入也很方便，不同于 ElasticJob 定义任务实现类，是通过@XxlJob 注解定义 JobHandler。

      1. @Component
      2. public class SampleXxlJob {
      3.     private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);
      4.     /**
      5.      * 1、简单任务示例（Bean模式）
      6.      */
      7.     @XxlJob("demoJobHandler")
      8.     public ReturnT demoJobHandler(String param) throws Exception {
      9.         XxlJobLogger.log("XXL-JOB, Hello World.");
      10. ​
      11. ​
      12.         for (int i = 0; i < 5; i++) {
      13.             XxlJobLogger.log("beat at:" + i);
      14.             TimeUnit.SECONDS.sleep(2);
      15.         }
      16.         return ReturnT.SUCCESS;
      17.     }
      18. ​
      19.     /**
      20.      * 2、分片广播任务
      21.      */
      22.     @XxlJob("shardingJobHandler")
      23.     public ReturnT shardingJobHandler(String param) throws Exception {
      24. ​
      25.         // 分片参数
      26.         ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
      27.         XxlJobLogger.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());
      28. ​
      29.         // 业务逻辑
      30.         for (int i = 0; i < shardingVO.getTotal(); i++) {
      31.             if (i == shardingVO.getIndex()) {
      32.                 XxlJobLogger.log("第 {} 片, 命中分片开始处理", i);
      33.             } else {
      34.                 XxlJobLogger.log("第 {} 片, 忽略", i);
      35.             }
      36.         }
      37.         return ReturnT.SUCCESS;
      38.     }
      39. }
      40. ​

    • XXL-JOB 相较于 ElasticJob，最大的特点就是功能比较丰富，可运维能力比较强，不但支持控制台动态创建任务，还有调度日志、运行报表等功能。

    •  

    • XXL-JOB 的历史记录、运行报表和调度日志，都是基于数据库实现的：

    • 由此可以看出，XXL-JOB 所有功能都依赖数据库，且调度中心不支持分布式架构，在任务量和调度量比较大的情况下，会有性能瓶颈。不过如果对任务量级、高可用、监控报警、可视化等没有过高要求的话，XXL-JOB 基本可以满足定时任务的需求。

  • 相关阅读:
    【教师资格证考试综合素质——法律专项】教育法笔记以及练习题
    MathType2024苹果版数学公式编辑器
    同步复位，异步复位，异步释放同步复位
    每日三题 9.06
    Spring框架笔记
    流媒体弱网优化之路(机器学习应用)——了解我们的网络模型
    (转帖)微服务拆分的原则和方法（2）
    C/C++常用函数
    PostgreSQL单机编译安装手册
    electron之进程间通信
  • 原文地址：https://blog.csdn.net/Andrew_Chenwq/article/details/127099951