Java基于DelayQueue实现周期任务实例

红太狼 2022-05-27 06:49 213阅读 0赞

## Executor结构图 ##

![这里写图片描述][70]

## 自定义周期执行任务 ##

首先看一下执行结果：

--- 开始执行任务 ---
    任务1：执行第1次，执行时间：13:29:49:945
    任务1：下次执行时间：13:30:06:769
    任务10：执行第1次，执行时间：13:29:49:945
    任务10：下次执行时间：13:29:52:595
    任务2：执行第1次，执行时间：13:29:49:945
    任务2：下次执行时间：13:30:04:773
    任务4：执行第1次，执行时间：13:29:49:945
    任务4：下次执行时间：13:29:56:745
    任务8：执行第1次，执行时间：13:29:49:945
    任务8：下次执行时间：13:29:52:425
    任务9：执行第1次，执行时间：13:29:49:945
    任务9：下次执行时间：13:30:04:793
    任务5：执行第1次，执行时间：13:29:49:945
    任务5：下次执行时间：13:30:01:791
    任务3：执行第1次，执行时间：13:29:49:945
    任务3：下次执行时间：13:29:51:645
    任务6：执行第1次，执行时间：13:29:49:945
    任务6：下次执行时间：13:29:51:173
    任务7：执行第1次，执行时间：13:29:49:945
    任务7：下次执行时间：13:30:09:117
    任务6：执行第2次，执行时间：13:29:51:173
    任务6：下次执行时间：13:30:15:776
    任务3：执行第2次，执行时间：13:29:51:645
    任务3：下次执行时间：13:30:16:671
    任务8：执行第2次，执行时间：13:29:52:425
    任务8：下次执行时间：13:30:00:324

代码：

import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    public class Main { 
    
        public static DelayQueue<MyDelayedTask> taskQueue = new DelayQueue<MyDelayedTask>();
    
        public static ExecutorService pool = Executors.newFixedThreadPool(5);
    
        public static void main(String args[]) {
            // 初始化队列，并启动添加任务的线程
            initTaskQueue();
            // 开始任务
            startTasks();
    
            synchronized (Main.class) {
                try {
                    Main.class.wait();
                } catch (InterruptedException e) {
                    System.out.println("同步代码块出错了");
                }
            }
        }
    
        // 开启线程
        public static void startTasks() {
            System.out.println("--- 开始执行任务 ---");
    
            pool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        while (true) {
                            // 每500ms执行一次
                            Thread.sleep(500);
                            MyDelayedTask task = taskQueue.take();
                            if (null != task) {
                                pool.execute(task);
                            }
                        }
                    } catch (InterruptedException e) {
                    }
                }
            });
    
        }
    
        // 初始化队列
        public static void initTaskQueue() {
    
            // 初始化 10 个任务
            for (int i = 1; i <= 10; i++) {
                taskQueue.put(new MyDelayedTask(new Date().getTime(), "任务" + i));
            }
            // 初始化
            pool.execute(new Runnable() {
                @Override
                public void run() {
                    int i = 11;
                    while (true) {
                        if (taskQueue.isEmpty()) {
                            taskQueue.put(new MyDelayedTask(new Date().getTime(), "任务" + i++));
                        }
                    }
                }
    
            });
    
        }
    
    }
    
    class MyDelayedTask implements Runnable, Delayed {
    
        // 执行次数
        private int exeTimes = 1;
        // 执行时间
        private long time;
        // 任务名称
        private String taskName;
    
        private Random random = new Random();
    
        public MyDelayedTask() {
        }
    
        public MyDelayedTask(long time, String taskName) {
            this.time = time;
            this.taskName = taskName;
        }
    
        /** * * @Title: compareTo * @Description: 按照时间比较排序 * @see java.lang.Comparable#compareTo(java.lang.Object) * @param o * @return */
        @Override
        public int compareTo(Delayed o) {
            if (this == o) {
                return 0;
            }
            if (o instanceof MyDelayedTask) {
                MyDelayedTask m = (MyDelayedTask) o;
                return this.time > m.time ? 1 : (this.time < m.time ? -1 : 0);
            }
    
            long d = (getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }
    
        /** * * @Title: getDelay * @Description: 计算时间间隔 * @see java.util.concurrent.Delayed#getDelay(java.util.concurrent.TimeUnit) * @param unit * @return */
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.time - System.currentTimeMillis(), TimeUnit.NANOSECONDS);
        }
    
        /** * * @Title: run * @Description: 执行任务，并生成下次执行的任务 * @see java.lang.Runnable#run() */
        @Override
        public void run() {
            System.out.println(taskName + "：执行第" + exeTimes + "次，执行时间：" + DateUtils.format(this.time, "HH:mm:ss:SSS"));
            // 记录次数
            exeTimes++;
            // 如果次数小于等于5，计算下次时间
            if (exeTimes <= 5) {
                // 下次执行时间
                this.time = getNextExecuteTime();
                System.out.println(taskName + "：下次执行时间：" + DateUtils.format(this.time, "HH:mm:ss:SSS"));
                Main.taskQueue.put(this);
            } else {
                System.out.println(taskName + "已结束");
            }
        }
    
        /** * * @Title: getNextExecuteTime * @Description: 计算下次执行时间 * @return */
        public long getNextExecuteTime() {
            return this.time + this.exeTimes * random.nextInt(10000);
        }
    }
    
    class DateUtils {
    
        private static Map<String, ThreadLocal<SimpleDateFormat>> sdfMap = new HashMap<String, ThreadLocal<SimpleDateFormat>>();
        private static final Object lockObj = new Object();
    
        private static SimpleDateFormat getSdf(final String pattern) {
            ThreadLocal<SimpleDateFormat> simpleDateFormat = sdfMap.get(pattern);
            // 此处的双重判断和同步是为了防止sdfMap这个单例被多次put重复的sdf
            if (simpleDateFormat == null) {
                synchronized (lockObj) {
                    simpleDateFormat = sdfMap.get(pattern);
                    if (simpleDateFormat == null) {
                        simpleDateFormat = new ThreadLocal<SimpleDateFormat>() {
                            @Override
                            protected SimpleDateFormat initialValue() {
                                return new SimpleDateFormat(pattern);
                            }
                        };
                        sdfMap.put(pattern, simpleDateFormat);
                    }
                }
            }
            return simpleDateFormat.get();
        }
    
        public static String format(Date date, String pattern) {
            return getSdf(pattern).format(date);
        }
    
        public static String format(long time, String pattern) {
            return format(new Date(time), pattern);
        }
    }

[70]: /images/20220527/05ce7c4b1f794c71902164de79ef936f.png