Quartz定时任务Redis集群方案

Bertha 。 2022-05-28 10:08 673阅读 0赞

加入jfinal club也有1个月之多了，学习到很多，总是向人家索取，自己并未付出的这种状态不是我想要的。本人实力有限也想尽可能的提供自己的绵薄之力，这次就抛砖引入分享一下Quartz定时任务Redis集群方案，参考社区之前有人已经提供了的QuartzPlugin和波总在cron4j中解答的如何去现实cron4j的集群方案

为什么要做redis集群方案

有人会说Quartz已经有一套成熟的集群方案了，但是那个是数据库版本，而且配置我也觉得挺麻烦，我不喜欢直连数据库，感觉很费资源

QuartzPlugin我就不贴出来了，之前就有人分享过

主要实现的接口，里面用到的一些redis操作是自己写的RedisCache，模仿jfinal ehcache的写法

主要流程

![blob.png][]

1.  `package com.jfinalshop.task;`
2.  ` `
3.  `import com.jfinal.ext2.kit.JsonExtKit;`
4.  `import com.jfinal.kit.JsonKit;`
5.  `import com.jfinal.kit.StrKit;`
6.  `import com.jfinalshop.util.IQuartzJobLoader;`
7.  `import com.jfinalshop.util.RedisCacheKit;`
8.  `import org.apache.commons.collections.map.HashedMap;`
9.  `import org.quartz.Job;`
10. `import org.quartz.JobExecutionContext;`
11. ` `
12. `import java.util.ArrayList;`
13. `import java.util.List;`
14. `import java.util.Map;`
15. `import java.util.concurrent.ConcurrentHashMap;`
16. `import java.util.concurrent.locks.Lock;`
17. `import java.util.concurrent.locks.ReentrantLock;`
18. ` `
19. `/**`
20. ` * Created by little fish on 2017/8/5.`
21. ` */`
22. ` `
23. `public abstract class RedisJob implements Job {`
24. `    private final String JOB_CACHE_PREFIX = "quartz_job_";`
25. `    private final String JOB_LOCK_KEY = "isLock";`
26. `    private final String JOB_RUN_COUNT_KEY = "runCount";`
27. `    private final String JOB_FAIL_COUNT_KEY = "failCount";`
28. `    private final String JOB_FAIL_MESSAGE_KEY = "failMessage";`
29. `    private final String JOB_LAST_FIRE_TIME_KEY = "lastFireTime";`
30. `    private final Integer JOB_LOCK_TRUE = 1;`
31. `    private final Integer JOB_LOCK_FALSE = 0;`
32. ` `
33. `    private static ConcurrentHashMap<String, ReentrantLock> lockMap = new ConcurrentHashMap<String, ReentrantLock>();`
34. ` `
35. `    private ReentrantLock getLock(String key) {`
36. `        ReentrantLock lock = lockMap.get(key);`
37. `        if (lock != null)`
38. `            return lock;`
39. ` `
40. `        lock = new ReentrantLock();`
41. `        ReentrantLock previousLock = lockMap.putIfAbsent(key, lock);`
42. `        return previousLock == null ? lock : previousLock;`
43. `    }`
44. ` `
45. `    private String genJobCacheName(JobExecutionContext jobExecutionContext){`
46. `        return String.format("%s%s", JOB_CACHE_PREFIX, jobExecutionContext.getJobInstance().getClass().getSimpleName());`
47. `    }`
48. ` `
49. `    private Map<Object, Object> genResetJobCacheMap(int runCount, Long fireTime,int failCount, String failMessage){`
50. `        Map<Object, Object> loadData = new HashedMap() ;`
51. `        loadData.put(JOB_LOCK_KEY, JOB_LOCK_FALSE);//去锁`
52. `        loadData.put(JOB_LAST_FIRE_TIME_KEY, fireTime);//同步最后一次触发时间`
53. `        loadData.put(JOB_RUN_COUNT_KEY, runCount++);//运行次数加1`
54. `        loadData.put(JOB_FAIL_COUNT_KEY, failCount);//错误次数`
55. `        loadData.put(JOB_FAIL_MESSAGE_KEY, failMessage);//错误信息`
56. `        return loadData;`
57. `    }`
58. ` `
59. `    public boolean checkJobValid(JobExecutionContext jobExecutionContext, IQuartzJobLoader jobLoader){`
60. `        String cacheName = genJobCacheName(jobExecutionContext);`
61. `        Integer isLock = RedisCacheKit.hget(cacheName, JOB_LOCK_KEY);`
62. `        if(isLock == JOB_LOCK_TRUE){`
63. `            return false;`
64. `        }`
65. ` `
66. `        Lock lock = getLock(cacheName);`
67. `        lock.lock();`
68. `        //初始化一些默认值`
69. `        Integer runCount = 0;`
70. `        Integer failCount = 0;`
71. `        Long lastFireTime = 0L;`
72. `        String failMessageJsonStr = "";`
73. `        //实际插入redis的数据`
74. `        Map<Object, Object> loadData = null;`
75. ` `
76. `        try {`
77. `            //再次拿出数据做double check检查并发`
78. `            Map<Object, Object> redisCacheData =  RedisCacheKit.hgetAll(cacheName);`
79. `            isLock = (Integer)redisCacheData.get(JOB_LOCK_KEY);`
80. `            lastFireTime = (Long)redisCacheData.get(JOB_LAST_FIRE_TIME_KEY);`
81. `            lastFireTime = lastFireTime != null?lastFireTime:0;`
82. `            runCount = (Integer)redisCacheData.get(JOB_RUN_COUNT_KEY);`
83. `            runCount = runCount != null?runCount:0;`
84. `            failCount = (Integer)redisCacheData.get(JOB_FAIL_COUNT_KEY);`
85. `            failCount = failCount != null?failCount:0;`
86. `            failMessageJsonStr = (String) redisCacheData.get(JOB_FAIL_MESSAGE_KEY);`
87. `            failMessageJsonStr = failMessageJsonStr != null?failMessageJsonStr:"";`
88. `            //检查是否满足执行条件`
89. `            if ((isLock == null || isLock == JOB_LOCK_FALSE) && lastFireTime < jobExecutionContext.getFireTime().getTime()) {`
90. `                //Redis加锁`
91. `                RedisCacheKit.hset(cacheName, JOB_LOCK_KEY, JOB_LOCK_TRUE);`
92. `                //处理Task主要业务逻辑`
93. `                jobLoader.load();`
94. `                //处理成功后把最新的状态同步到Redis中`
95. `                loadData = genResetJobCacheMap(JOB_LOCK_FALSE, jobExecutionContext.getFireTime().getTime(), failCount, failMessageJsonStr);`
96. `            }`
97. ` `
98. `            return true;`
99. `        }catch (Exception e){`
100. `            e.printStackTrace();`
101. `            failCount++;//失败次数+1`
102. ` `
103. `            //添加失败信息以便之后检查`
104. `            List<RedisJobFailMessage> faileMessageArray = null;`
105. `            if(StrKit.isBlank(failMessageJsonStr)){`
106. `                faileMessageArray = new ArrayList<RedisJobFailMessage>();`
107. `            }else{`
108. `                try {`
109. `                    faileMessageArray = JsonExtKit.jsonToJSONArray(failMessageJsonStr).toJavaList(RedisJobFailMessage.class);`
110. `                }catch (Exception ex){`
111. `                    ex.printStackTrace();`
112. `                }`
113. ` `
114. `                if(faileMessageArray == null){`
115. `                    faileMessageArray = new ArrayList<RedisJobFailMessage>();`
116. `                }`
117. `            }`
118. ` `
119. `            faileMessageArray.add(new RedisJobFailMessage(lastFireTime, e.getMessage()));`
120. `            failMessageJsonStr = JsonKit.toJson(faileMessageArray);`
121. `            loadData = genResetJobCacheMap(JOB_LOCK_FALSE, jobExecutionContext.getFireTime().getTime(), failCount, failMessageJsonStr);`
122. ` `
123. `        }finally {`
124. `            RedisCacheKit.hmsetForever(cacheName, loadData);`
125. `            lock.unlock();`
126. `        }`
127. ` `
128. `        return false;`
129. `    }`
130. `}`

`一个Loader接口其实就是和IDataLoader一模一样`

1.  package com.jfinalshop.util;
2.   
3.  /\*\*
4.   \* Created by little fish on 2017/7/16.
5.   \*/
6.  public interface IQuartzJobLoader \{ 
7.      public void load(Object... args);
8.  \}

最后调用

1.  package com.jfinalshop.task;
2.   
3.  import com.jfinal.kit.LogKit;
4.  import com.jfinalshop.util.IQuartzJobLoader;
5.  import org.quartz.JobExecutionContext;
6.  import org.quartz.JobExecutionException;
7.   
8.  /\*\*
9.   \* Created by little fish on 2017/8/5.
10.  \*/
11. public class TaskTest extends RedisJob \{ 
12.     @Override
13.     public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException \{ 
14.         this.checkJobValid(jobExecutionContext, new IQuartzJobLoader() \{ 
15.             @Override
16.             public void load(Object... args) \{ 
17.                 //验证通过执行任务
18.                 LogKit.info(String.format("getClass = %s fireTime = %s runtime = %s nextFireTime = %s result = %s",jobExecutionContext.getJobInstance().getClass().getSimpleName(),jobExecutionContext.getFireTime(),jobExecutionContext.getJobRunTime(),jobExecutionContext.getNextFireTime(),jobExecutionContext.getResult()));
19.             \}
20.         \});
21.  
22.     \}
23. \}

刚写完的，代码写的很粗糙，只是做过冒烟测试，可以正常执行，没有做过优化

[blob.png]: /images/20220528/b4d6e5932dd7471da8e1d3b9a9175465.png