Quartz超时重试机制

115 阅读 0 评论 76 点赞

我是靠谱客的博主仁爱黑猫，这篇文章主要介绍Quartz超时重试机制，现在分享给大家，希望可以做个参考。

高可用作为考究系统的一项重要指标，如何做到系统的高可用，谈及一个系统，这个话题就难以越过。Quartz作为目前调度框架的一个流行组件，如何保证Quartz的高可用，任务调度失败后，如何进行重试，这个也是一个值得关注的问题。

网上看过许多涉及定时调度的开源项目，但发现其都存在一个问题，并未对任务调度的失败做处理，仅仅只是简单的日志记录，以及手工重采。但是现实情况中，往往可能因为间歇性原因导致调度的失败，可能重新调度一下就可以成功，那么这时失败重试就尤为重要。

找过网上关于Quartz失败后如何实现重试。倒是找到过集中方式。

引入第三方jar包，spring-retry。可以通过注解的方式或者集成对应的类，重写对应的方法。个人观点。又引入第三方jar，感觉不太理性。同时感觉过于复杂，有的业务需求不太好控制。
在Quartz里面抛异常时，自己用线程类睡多长时间后，进行再次调用。这种存在很大缺点。线程一直阻塞，不太优雅。当然可以用线程池的方式来进行实现，这种我没尝试过。
Quartz抛异常时，获取失败的job，设置启动时间，策略执行一次，这种方式，个人觉得最好，够优雅

下面简单介绍一下这种方式，直接贴源码：

复制代码

public abstract class AbstractQuartzJob implements Job
{
    private static final Logger log = LoggerFactory.getLogger(AbstractQuartzJob.class);

@Override
    public void execute(JobExecutionContext context) {
        ParamMap<String, Object> sysJob = (ParamMap<String, Object>) context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES);

try
        {
            JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
            if(!jobDataMap.containsKey("currRetryCount")) {
                before(context, sysJob);
            }
            if (sysJob != null)
            {
                doExecute(context, sysJob);
            }
            afterReturning(context, sysJob);
        } catch (Exception e)
        {
            afterThrowing(context, e, sysJob);
        }
    }

/**
     * @description
     *      在抛出异常后
     * @Date        2019/6/28 15:57
     * @Author      wangzx
     * @Version     V1.0.20
     **/
    protected void afterThrowing(JobExecutionContext context, Exception e, ParamMap<String, Object> sysJob) {
        JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
        int maxRetryCount = Integer.parseInt(YmlUtils.getValue("schedule.maxAttempt").toString());
        long retryInterval = Long.parseLong(YmlUtils.getValue("schedule.delay")) * 60 * 1000;
        ParamMap<String, Object> paramMap = new ParamMap<>();

int currRetryCount = jobDataMap.containsKey("currRetryCount") ? jobDataMap.getIntValue("currRetryCount") : 0;
        if(currRetryCount < maxRetryCount) {
            jobDataMap.put("currRetryCount", currRetryCount + 1);

JobDetail job = context
                    .getJobDetail()
                    .getJobBuilder()
                    .withIdentity(context.getJobDetail().getKey().getName() + "_" + currRetryCount, "FailingJobsGroup")
                    .usingJobData(jobDataMap)
                    .build();

OperableTrigger trigger = (OperableTrigger) TriggerBuilder
                    .newTrigger()
                    .forJob(job)
                    .startAt(new Date(context.getFireTime().getTime() + retryInterval))
                    .build();

try {
                context.getScheduler().scheduleJob(job, trigger);
            } catch (SchedulerException e1) {
                e1.printStackTrace();
            }
            paramMap.put("motTaskStatus", "1");
            paramMap.put("motTaskId", sysJob.get("motTaskId"));
            SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
        } else {
            log.error("任务执行异常:", e);
            after(context, e, sysJob);

paramMap.put("motTaskStatus", "3");
            String currDateStr = DateUtil.format(new Date(), "yyyyMMdd HHmmss");
            String[] strs = currDateStr.split(" ");
            String beginTime = strs[1];
            paramMap.put("endTime", beginTime);
            paramMap.put("motTaskId", sysJob.get("motTaskId"));
            // 写入数据库当中
            SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
        }
    }

/**
     * @description
     *      返回后
     * @Date        2019/6/28 16:01
     * @Author      wangzx
     * @Version     V1.0.20
     **/
    protected void afterReturning(JobExecutionContext context, ParamMap<String, Object> sysJob) {
        String currDateStr = DateUtil.format(new Date(), "yyyyMMdd HHmmss");
        String[] strs = currDateStr.split(" ");
        String beginTime = strs[1];

ParamMap<String, Object> paramMap = new ParamMap<>();
        paramMap.put("endTime", beginTime);
        paramMap.put("motTaskStatus", "2");
        paramMap.put("motTaskId", sysJob.get("motTaskId"));

// 写入数据库当中
        SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
    }

/**
     * 执行前
     *
     * @param context 工作执行上下文对象
     * @param sysJob 系统计划任务
     */
    protected void before(JobExecutionContext context, ParamMap<String, Object> sysJob)
    {
        String currDateStr = DateUtil.format(new Date(), "yyyyMMdd HHmmss");
        String[] strs = currDateStr.split(" ");
        String currDate = strs[0];
        String beginTime = strs[1];
        ParamMap<String, Object> paramMap = new ParamMap<>();
        paramMap.put("currDate", currDate);
        paramMap.put("beginTime", beginTime);
        paramMap.put("motTaskId", sysJob.get("motTaskId"));

// 写入数据库当中
        SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
    }

/**
     * 执行后
     *
     * @param context 工作执行上下文对象
     */
    protected void after(JobExecutionContext context, Exception e, ParamMap<String, Object> sysJob) {
    }

/**
     * 执行方法，由子类重载
     *
     * @param context 工作执行上下文对象
     * @param sysJob 系统计划任务
     * @throws Exception 执行过程中的异常
     */
    protected abstract void doExecute(JobExecutionContext context, ParamMap<String, Object> sysJob) throws Exception;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
public abstract class AbstractQuartzJob implements Job
{
    private static final Logger log = LoggerFactory.getLogger(AbstractQuartzJob.class);

    @Override
    public void execute(JobExecutionContext context) {
        ParamMap<String, Object> sysJob = (ParamMap<String, Object>) context.getMergedJobDataMap().get(ScheduleConstants.TASK_PROPERTIES);

        try
        {
            JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
            if(!jobDataMap.containsKey("currRetryCount")) {
                before(context, sysJob);
            }
            if (sysJob != null)
            {
                doExecute(context, sysJob);
            }
            afterReturning(context, sysJob);
        } catch (Exception e)
        {
            afterThrowing(context, e, sysJob);
        }
    }

    /**
     * @description
     *      在抛出异常后
     * @Date        2019/6/28 15:57
     * @Author      wangzx
     * @Version     V1.0.20
     **/
    protected void afterThrowing(JobExecutionContext context, Exception e, ParamMap<String, Object> sysJob) {
        JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
        int maxRetryCount = Integer.parseInt(YmlUtils.getValue("schedule.maxAttempt").toString());
        long retryInterval = Long.parseLong(YmlUtils.getValue("schedule.delay")) * 60 * 1000;
        ParamMap<String, Object> paramMap = new ParamMap<>();

        int currRetryCount = jobDataMap.containsKey("currRetryCount") ? jobDataMap.getIntValue("currRetryCount") : 0;
        if(currRetryCount < maxRetryCount) {
            jobDataMap.put("currRetryCount", currRetryCount + 1);

            JobDetail job = context
                    .getJobDetail()
                    .getJobBuilder()
                    .withIdentity(context.getJobDetail().getKey().getName() + "_" + currRetryCount, "FailingJobsGroup")
                    .usingJobData(jobDataMap)
                    .build();

            OperableTrigger trigger = (OperableTrigger) TriggerBuilder
                    .newTrigger()
                    .forJob(job)
                    .startAt(new Date(context.getFireTime().getTime() + retryInterval))
                    .build();

            try {
                context.getScheduler().scheduleJob(job, trigger);
            } catch (SchedulerException e1) {
                e1.printStackTrace();
            }
            paramMap.put("motTaskStatus", "1");
            paramMap.put("motTaskId", sysJob.get("motTaskId"));
            SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
        } else {
            log.error("任务执行异常:", e);
            after(context, e, sysJob);

            paramMap.put("motTaskStatus", "3");
            String currDateStr = DateUtil.format(new Date(), "yyyyMMdd HHmmss");
            String[] strs = currDateStr.split(" ");
            String beginTime = strs[1];
            paramMap.put("endTime", beginTime);
            paramMap.put("motTaskId", sysJob.get("motTaskId"));
            // 写入数据库当中
            SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
        }
    }

    /**
     * @description
     *      返回后
     * @Date        2019/6/28 16:01
     * @Author      wangzx
     * @Version     V1.0.20
     **/
    protected void afterReturning(JobExecutionContext context, ParamMap<String, Object> sysJob) {
        String currDateStr = DateUtil.format(new Date(), "yyyyMMdd HHmmss");
        String[] strs = currDateStr.split(" ");
        String beginTime = strs[1];

        ParamMap<String, Object> paramMap = new ParamMap<>();
        paramMap.put("endTime", beginTime);
        paramMap.put("motTaskStatus", "2");
        paramMap.put("motTaskId", sysJob.get("motTaskId"));

        // 写入数据库当中
        SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
    }

    /**
     * 执行前
     *
     * @param context 工作执行上下文对象
     * @param sysJob 系统计划任务
     */
    protected void before(JobExecutionContext context, ParamMap<String, Object> sysJob)
    {
        String currDateStr = DateUtil.format(new Date(), "yyyyMMdd HHmmss");
        String[] strs = currDateStr.split(" ");
        String currDate = strs[0];
        String beginTime = strs[1];
        ParamMap<String, Object> paramMap = new ParamMap<>();
        paramMap.put("currDate", currDate);
        paramMap.put("beginTime", beginTime);
        paramMap.put("motTaskId", sysJob.get("motTaskId"));

        // 写入数据库当中
        SpringUtils.getBean(F051_F100_IMotTaskDao.class).f2106054(paramMap);
    }

    /**
     * 执行后
     *
     * @param context 工作执行上下文对象
     */
    protected void after(JobExecutionContext context, Exception e, ParamMap<String, Object> sysJob) {
    }

    /**
     * 执行方法，由子类重载
     *
     * @param context 工作执行上下文对象
     * @param sysJob 系统计划任务
     * @throws Exception 执行过程中的异常
     */
    protected abstract void doExecute(JobExecutionContext context, ParamMap<String, Object> sysJob) throws Exception;
}