多线程应用：定时任务重发机制实现

痛定思痛。 2023-02-19 07:29 6阅读 0赞

> 项目场景：对于定时任务发起的一些任务，由于某些客观原因，导致任务失败，希望能够有一个监听的定时器，定时轮训任务状态表，并重新发起任务调用。

> 核心思路：  
> 1.定义一个定时器TaskARQCron.java，发起失败任务扫描。  
> 2.定义一张新的重发状态表，区分开原任务状态表，每重新发起一次任务，重发状态表记录一条数据。  
> 3.智能识别依赖性任务和非依赖性任务  
> 4.定时任务的扫描的间隔时间设置 2个小时  
> 5.按照日期先后顺序执行重发,先失败,先执行原则,同时针对非依赖性任务,排序执行  
> 6.对于已选中的重发任务，使用ExecutorService、CountDownLatch实现多线程并发执行。  
> 7.将最大重试次数，定义到配置文件，或者任务枚举类中。

## 第一步：定时任务表达式 ##

# 每个小时执行一次
    cronExpression: 0 0 0/2 * * ?

关于cron表达式的定义，可参考: [https://www.cnblogs.com/javahr/p/8318728.html][https_www.cnblogs.com_javahr_p_8318728.html]

## 第二步：定时器类 ##

public class TaskARQCron implements Runnable { 
    	public void run() { 
    		ITaskARQ taskARQ = (ITaskARQ) (ApplicationContextHelper.applicationContext.getBean("taskARQClient"));
            taskARQResp = taskARQ.exe();
    	}
    }

## 第三步：任务处理接口Service ##

public interface ITaskARQ { 
        BaseSvcResp exe();
    }

## 第四步：任务处理Service实现 ##

public class TaskARQService implements ITaskARQ { 
        /** 执行时间到达时, 所有的线程需要依次退出, 主线程才开始统计执行事物总数 */
        private static CountDownLatch countDownLatch;
        private ExecutorService executorService;
        
    	public BaseSvcResp exe() { 
    		//1.初始化线程池大小
    		executorService = Executors.newFixedThreadPool(poolSize);
    		//2.查询满足条件的记录数
    		List<TblPipbatBatSt> filter = batStMapper.selectByWhiteList(allowList);
    		//3.初始化CountDownLatch容量大小,用于统计并发线程的结束的状态
    		countDownLatch = new CountDownLatch(ableList.size());
    		//4.根据任务数量,循环创建重发任务线程去执行
    		for (TblPipbatBatSt batSt : ableList) { 
    			TaskEvent taskEvent = new TaskEvent();
    			//TODO ... 构建taskEvent对象
    			executorService.submit(new Worker(job, taskEvent));
    		}
    		//5.等待所有线程结束
    		try { 
                countDownLatch.await();
                logger.info("所有线程执行结束-----------------");
            } catch (InterruptedException e) { 
                //do something....
            }
            //6.关闭资源
            executorService.shutdown();
    	}
    }

## 第五步：定义Job、JobDetail模式(内部类) ##

interface Job{ 
            BaseSvcResp execute(TaskEvent taskEvent);
        }
    
        static class JobDetail implements Job{ 
            @Override
            public BaseSvcResp execute(TaskEvent taskEvent) { 
                //TODO ... 重新执行任务调用
                logger.info("任务重发成功,当前线程:{},当前任务:{}",Thread.currentThread().getName(),taskEvent.toString());
                return resp;
            }
        }

## 第六步：定义线程类Worker(内部类) ##

class Worker implements Runnable { 
    	private Job job;
        private TaskEvent taskEvent;
    
        Worker (Job job, TaskEvent taskEvent) { 
        	this.job = job;
            this.taskEvent = taskEvent;
        }
    	public void run() { 
    		//do something need ...
    		BaseSvcResp result = this.job.execute(taskEvent);
    		//do something need ...
    		countDownLatch.countDown();//代表子线程任务结束
    	}
    }

> 学习Java的同学注意了！！！  
> 学习过程中遇到什么问题或者想获取学习资源的话，欢迎加入Java学习交流群，群号码：543120397 我们一起学Java！

[https_www.cnblogs.com_javahr_p_8318728.html]: https://www.cnblogs.com/javahr/p/8318728.html