ThreadPoolTaskScheduler手写任务调度

先贴一个自己写的demo把，原理其实就是这样的。

先记录一个东西，后续来研究：CronSequenceGenerator

报错：Cron expression must consist of 6 fields，这玩意支持6位不支持7位，真是XXXX，quartz支持，参见 https://my.oschina.net/uwith/blog/4395339 。

CronTrigger这个类可以将cron表达式转换成Date，可以查看schedule源码学到不少东西，下面代码就是转换成下一执行时间。

public Date nextExecutionTime(TriggerContext triggerContext)

@Slf4j @RestController public class HomeController {

_/\*\*

_ * 存储调度器信息_,存在多线程安全问题,__采用__ConcurrentHashMap _ * _初始大小设定大于核心线程池数量__3__倍即可 _ _*/ _ private Map<Integer, ScheduledFuture> map = new ConcurrentHashMap<>(64); ThreadPoolTaskScheduler threadPoolTaskScheduler; public HomeController() { // 注意，此线程池使用DelayedWorkQueue()作为队列，此队列又是基于PriorityQueue, // 自己会根据表达式进行排序,可以查看schedule方法源码以及参数 threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); threadPoolTaskScheduler.initialize(); // 线程池需做配置化 threadPoolTaskScheduler.setPoolSize(64); }

1@RequestMapping("/home/index/{id}")
2public String add(@PathVariable Integer id, @RequestBody String cron) {
3    // 添加一个去重复的操作

if (map.containsKey(id)) { return "调度任务已添加成功"; } ScheduledFuture future = threadPoolTaskScheduler.schedule( new Runnable() { @Override public void run() { try { // 触发进行记录日志 StopWatch watch = new StopWatch(); watch.start(); log.info(String.format("%s-调度器已触发,准备rpc调用", id)); // 这里执行具体的操作 Thread.sleep(1000); //执行完成取消 Boolean cancelResult = cancel(id); watch.stop(); log.info(String.format("%s-调度器rpc调用完成,耗时:%s", id, watch.getLastTaskTimeMillis())); } catch (Exception ex) { log.error(String.format("%s-调度器出现异常:%s,堆栈信息:%s", id, ex.getMessage(), ExceptionUtils.getStackTrace(ex))); } } }, new CronTrigger(cron)); // 暂时缓存，用于控制 map.put(id, future); return "调度任务已添加成功"; }

1@RequestMapping("/home/cancel/{id}")
2public Boolean cancel(@PathVariable Integer id) {
3    ScheduledFuture future = map.get(id);

// mayInterruptIfRunning参数指明是否可以中断线程,         // 注意如果写true那么该段代码执行完成，后续操作可能会被中断，比如下面的remove方法，可能执行不到 Boolean result = future.cancel(true); // 然后移除缓存 map.remove(id); return Boolean.TRUE; }

1@RequestMapping("/home/query/{id}")
2public String query(@PathVariable Integer id) {
3    ScheduledFuture future = map.get(id);

StringBuilder sb = new StringBuilder(); sb.append(String.format("future.isCancelled：%s</br>", future.isCancelled())); sb.append(String.format("future.isDone：%s</br>", future.isDone())); return sb.toString(); }

1@RequestMapping("/home/queryInfo")
2public String queryInfo() {
3    StringBuilder sb = new StringBuilder();

sb.append(String.format("getPoolSize：%s</br>", threadPoolTaskScheduler.getPoolSize())); sb.append(String.format("getActiveCount：%s</br>", threadPoolTaskScheduler.getActiveCount())); sb.append(String.format("getScheduledExecutor.isShutdown", threadPoolTaskScheduler.getScheduledExecutor().isShutdown())); sb.append(String.format("getScheduledExecutor.isTerminated：%s</br>", threadPoolTaskScheduler.getScheduledExecutor().isTerminated())); sb.append(String.format("getScheduledThreadPoolExecutor.getPoolSize：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getPoolSize())); sb.append(String.format("getScheduledThreadPoolExecutor.getCorePoolSize：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getCorePoolSize())); sb.append(String.format("getScheduledThreadPoolExecutor.getLargestPoolSize：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getLargestPoolSize())); sb.append(String.format("getScheduledThreadPoolExecutor.getMaximumPoolSize：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getMaximumPoolSize())); sb.append(String.format("getScheduledThreadPoolExecutor.getActiveCount：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getActiveCount())); sb.append(String.format("getScheduledThreadPoolExecutor.getTaskCount：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getTaskCount())); sb.append(String.format("getScheduledThreadPoolExecutor.getCompletedTaskCount：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getCompletedTaskCount())); sb.append(String.format("getScheduledThreadPoolExecutor.getQueue：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getQueue().size())); sb.append(String.format("getScheduledThreadPoolExecutor.getKeepAliveTime：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().getKeepAliveTime(TimeUnit.SECONDS))); sb.append(String.format("getScheduledThreadPoolExecutor.isShutdown：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().isShutdown())); sb.append(String.format("getScheduledThreadPoolExecutor.isTerminating：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().isTerminating())); sb.append(String.format("getScheduledThreadPoolExecutor.isTerminated：%s</br>", threadPoolTaskScheduler.getScheduledThreadPoolExecutor().isTerminated())); return sb.toString(); }

}