您现在的位置是:网站首页> 内容页

java实现任务调度

  • 博猫游戏手机版 登录首页
  • 2019-09-27
  • 405人已阅读
简介最近的一个小项目是做一个简单的数据仓库,需要将其他数据库的数据抽取出来,并通过而出抽取成页面需要的数据,以空间换时间的方式,让后端报表查询更快。因为在抽取的过程中,有一定的先后顺序,需

最近的一个小项目是做一个简单的数据仓库,需要将其他数据库的数据抽取出来,并通过而出抽取成页面需要的数据,以空间换时间的方式,让后端报表查询更快。因为在抽取的过程中,有一定的先后顺序,需要做一个任务调度器,某一优先级的会先执行,然后会进入下一个优先级的队列任务中。先定义了一个Map的集合,key是优先级,value是任务的集合,某一个优先级内的任务是并发执行的,而不同优先级是串行执行的,前一个优先级执行完之后,后面的才会执行。

ConcurrentHashMap<Integer/* 优先级. */, List<BaseTask>/* 任务集合. */> tasks = new ConcurrentHashMap<>();

这个调度管理有一个演进的过程,我先说第一个,这个是比较好理解的。第一个版本:首先对tasks集合中的key进行一个排序,我定义的是数字越小就有限执行,则进行遍历key值,并取出某个优先级的任务队列,执行任务队列的任务。任务的执行交给线程池去执行,在遍历内部,需要不断的检查这个队列中的任务是否都执行了,没有则一直等待否则进入到下个队列,任务执行的时候可能会抛出异常,但是不管任务是否异常,都将任务状态设置已执行。下面是其核心代码:

public void run() { //对key值进行排序 Enumeration<Integer> keys = tasks.keys(); List<Integer> prioritys = new ArrayList<>(); while (keys.hasMoreElements()) { prioritys.add(keys.nextElement()); } Collections.sort(prioritys);//升序 //对key进行遍历,执行某个某个优先级的任务队列 for (Integer priority : prioritys) { List<BaseTask> taskList = tasks.get(priority); if (taskList.isEmpty()) { continue; } logger.info("execute priority {} task ", taskList.get(0).priority); for (BaseTask task : taskList) { executor.execute(() -> { try { task.doTask(); } catch (Exception e) { e.printStackTrace(); } });//线程中执行任务 } while (true) {//等待所有线程都执行完成之后执行下一个任务队列 boolean finish = true; for (BaseTask t : taskList) { if (!t.finish) { finish = false; } } if (finish) {//当前任务都执行完毕 break; } Misc.sleep(1000);//Thread.sleep(1000) } Misc.sleep(1000); } }

关键代码很好理解,在任务执行之前,需要对所有任务都初始化,初始化的时候给出每个任务的优先级和任务名称,任务抽象类如下:

public abstract class BaseTask { public String taskName;//任务名称 public Integer priority; //优先级 public boolean finish; //任务完成? /** * 执行的任务 */ public abstract void doTask(Date date) throws Exception;

第一个版本的思路很简单。第二个版本稍微有一点点复杂。这里主要介绍该版本的内容,后续将代码的链接附上。程序是由SpringBoot搭建起来的,定时器是Spring内置的轻量级的Quartz,使用Aop方式拦截异常,使用注解的方式在任务初始化时设置任务的初始变量。使用EventBus解耦程序,其中程序简单实现邮件发送功能(该功能还需要自己配置参数),以上这些至少需要简单的了解一下。程序的思路:在整个队列执行过程中会有多个管道,某个队列上的管道任务执行完成,可以直接进行到下一个队列中执行,也设置了等待某一个队列上的所有任务都执行完成才执行当前任务。在某个队列任务中会标识某些任务是一队的,其他的为另一队,当这一队任务执行完成,就可以到下一个队列中去,不需要等待另一队。这里会先初始化每个队列的每个队的条件,这个条件就是每个队的任务数,执行完成减1,当为0时,就进入下一个队列中。分四个步骤进行完成:1.bean的初始化2.条件的设置3.任务的执行4.任务异常和任务执行完成之后通知检查是否执行下一个队列的任务

1.bean的初始化

1.创建注解类

@Retention(RetentionPolicy.RUNTIME)@Target(ElementType.TYPE)@Documentedpublic @interface TaskAnnotation { int priority() default 0;//优先级 String taskName() default "";//任务名称 TaskQueueEnum[] queueName() default {};//队列名称}

2.实现BeanPostProcessor,该接口是中有两个方法postProcessBeforeInitialization和postProcessAfterInitialization,分别是bean初始化之前和bean初始化之后做的事情。

Annotation[] annotations = bean.getClass().getAnnotations();//获取类上的注解 if (ArrayUtils.isEmpty(annotations)) {//注解为空时直接返回(不能返回空,否则bean不会被加载) return bean; } for (Annotation annotation : annotations) { if (annotation.annotationType().equals(TaskAnnotation.class)) { TaskAnnotation taskAnnotation = (TaskAnnotation) annotation;//强转 try { Field[] fields = target.getClass().getFields();//需要通过反射将值进行修改,下面的操作仅仅是对象的引用 if (!ArrayUtils.isEmpty(fields)) { for (Field f : fields) { f.setAccessible(true); if (f.getName().equals("priority")) { f.set(target, taskAnnotation.priority()); } } } } }

上面需要注意的一点是需要通过反射的机制给bean设置值,不能直接调用bean的方式set值,否则bean的值是空的。上面的代码通过实现BeanPostProcessor后置处理器,处理任务上的注解,完成对任务的初始化的。

2.条件的初始化

创建条件类,提供初始化的方法。

public abstract class BaseTask { public int nextPriority;//子级节点的优先级 public String taskName;//任务名称 public Integer priority; //优先级 public String queueName;//队列名称 public boolean finish; //任务完成? public boolean allExecute; /** * 执行的任务 */ public abstract void doTask(Date date) throws Exception; //任务完成之后,通过eventBus发送通知,是否需要执行下一个队列 public void notifyExecuteTaskMsg(EventBus eventBus, Date date) { EventNotifyExecuteTaskMsg msg = new EventNotifyExecuteTaskMsg(); msg.setDate(date); msg.setNextPriority(nextPriority); msg.setQueueName(queueName); msg.setPriority(priority); msg.setTaskName(taskName); eventBus.post(msg); }}public class TaskExecuteCondition { private ConcurrentHashMap<String, AtomicInteger> executeMap = new ConcurrentHashMap<>(); /** * 初始化,每个队列进行分组,每个组的任务数量放入map集合中. */ public void init(ConcurrentHashMap<Integer, List<BaseTask>> tasks) { Enumeration<Integer> keys = tasks.keys(); List<Integer> prioritys = new ArrayList<>(); while (keys.hasMoreElements()) { prioritys.add(keys.nextElement()); } Collections.sort(prioritys);//升序 for (Integer priority : prioritys) { List<BaseTask> list = tasks.get(priority); if (list.isEmpty()) { continue; } //对每个队列进行分组 Map<String, List<BaseTask>> collect = list.stream() .collect(Collectors.groupingBy(x -> x.queueName, Collectors.toList())); for (Entry<String, List<BaseTask>> entry : collect.entrySet()) { for (BaseTask task : entry.getValue()) { addCondition(task.priority, task.queueName); } } } } /** * 执行任务完成,条件减1 */ public boolean executeTask(Integer priority, String queueName) { String name = this.getQueue(priority, queueName); AtomicInteger count = executeMap.get(name); int sum = count.decrementAndGet(); if (sum == 0) { return true; } return false; } /** * 对个某个队列的条件 */ public int getCondition(Integer priority, String queueName) { String name = this.getQueue(priority, queueName); return executeMap.get(name).get(); } private void addCondition(Integer priority, String queueName) { String name = this.getQueue(priority, queueName); AtomicInteger count = executeMap.get(name); if (count == null) { count = new AtomicInteger(0); executeMap.put(name, count); } count.incrementAndGet(); } private void addCondition(Integer priority, String queueName, int sum) { String name = this.getQueue(priority, queueName); AtomicInteger count = executeMap.get(name); if (count == null) { count = new AtomicInteger(sum); executeMap.put(name, count); } else { count.set(sum); } } private String getQueue(Integer priority, String queueName) { return priority + queueName; } /** * 清除队列 */ public void clear() { this.executeMap.clear(); }}

3.任务的执行

任务执行类提供run方法,执行第一个队列,并提供获取下一个队列优先级方法,执行某个队列某个组的方法。

public class ScheduleTask { private static final Logger logger = LoggerFactory.getLogger(ScheduleTask.class); public ConcurrentHashMap<Integer/* 优先级. */, List<BaseTask>/* 任务集合. */> tasks = new ConcurrentHashMap<>(); @Autowired private ThreadPoolTaskExecutor executor;//线程池 //任务会先执行第一队列的任务. public void run(Date date) { Enumeration<Integer> keys = tasks.keys(); List<Integer> prioritys = new ArrayList<>(); while (keys.hasMoreElements()) { prioritys.add(keys.nextElement()); } Collections.sort(prioritys);//升序 Integer priority = prioritys.get(0); executeTask(priority, date);//执行第一行的任务. } //获取下一个队列的优先级 public Integer nextPriority(Integer priority) { Enumeration<Integer> keys = tasks.keys(); List<Integer> prioritys = new ArrayList<>(); while (keys.hasMoreElements()) { prioritys.add(keys.nextElement()); } Collections.sort(prioritys);//升序 for (Integer pri : prioritys) { if (priority < pri) { return pri; } } return null;//没有下一个队列 } public void executeTask(Integer priority) { List<BaseTask> list = tasks.get(priority); if (list.isEmpty()) { return; } for (BaseTask task : list) { execute(task); } } //执行某个队列的某个组 public void executeTask(Integer priority, String queueName) { List<BaseTask> list = this.tasks.get(priority); list = list.stream().filter(task -> queueName.equals(task.queueName)) .collect(Collectors.toList()); if (list.isEmpty()) { return; } for (BaseTask task : list) { execute(task); } } public void execute(BaseTask task) { executor.execute(() -> { try { task.doTask(date);// } catch (Exception e) {//异常处理已经Aop拦截处理 } });//线程中执行任务 } /** * 增加任务 */ public void addTask(BaseTask task) { List<BaseTask> baseTasks = tasks.get(task.priority); if (baseTasks == null) { baseTasks = new ArrayList<>(); List<BaseTask> putIfAbsent = tasks.putIfAbsent(task.priority, baseTasks); if (putIfAbsent != null) { baseTasks = putIfAbsent; } } baseTasks.add(task); } /** * 将任务结束标识重新设置 */ public void finishTask() { tasks.forEach((key, value) -> { for (BaseTask task : value) { task.finish = false; } }); }}

4.任务异常和任务执行完成之后通知检查是否执行下一个队列的任务

public class EventNotifyExecuteTaskListener { private static final Logger logger = LoggerFactory .getLogger(EventNotifyExecuteTaskListener.class); @Autowired private ScheduleTask scheduleTask; @Autowired private TaskExecuteCondition condition; @Subscribe public void executeTask(EventNotifyExecuteTaskMsg msg) { //当前队列的某组内容是否都执行完成 boolean success = condition.executeTask(msg.getPriority(), msg.getQueueName()); if (success) { Integer nextPriority = scheduleTask.nextPriority(msg.getPriority()); if (nextPriority != null) { scheduleTask.executeTask(nextPriority, msg.getQueueName(), msg.getDate());//执行下一个队列 } else {//执行完成,重置任务标识 scheduleTask.finishTask(); logger.info("CoreTask end!"); } } }}

整个思路介绍到这里,那么接下来是整个项目中出现的一些问题1.BeanPostProcessor与Aop一起使用时,postProcessAfterInitialization调用之后获取的bean分为不同的了,一个是jdk原生实体对象,一种是Aop注解下的类会被cglib代理,生成带有后缀的对象,如果通过这个对象时反射获取类的注解,字段和方法,就获取不到,在代码中,需要将其转化一下,将cgLib代理之后的类转化为不带后缀的对象。2.postProcessAfterInitialization的参数bean不能直接设置值,就是如下:

TaskAnnotation taskAnnotation = (TaskAnnotation) annotation;//强转 BaseTask baseTask = (BaseTask) bean;//强转 baseTask.priority = taskAnnotation.priority();

在使用对象时,其中对象的字段时为空的,并需要通过反射的方式去设置字段的值。上面仅仅只是个人的想法,如果有更好的方式,或者有某些地方可以进行改进的,我们可以共同探讨一下。

链接地址:https://github.com/wangice/task-scheduler程序中使用了一个公共包:https://github.com/wangice/misc

文章评论

Top