Java多线程之等待队列DelayQueue详解

class DelayedTask implements Runnable, Delayed { // Delayed接口必须实现，Runnable接口可以不实现
    private static int counter = 0;
    private final int id = counter++;
    /**
     * 延迟的时间(单位: 毫秒)
     */
    private final int delta;
    /**
     * 任务准备执行的时间点(单位: 纳秒)
     */
    private final long trigger;
    protected static List<DelayedTask> sequence = new ArrayList<>();
    public DelayedTask(int delayInMilliseconds) {
        delta = delayInMilliseconds;
        trigger = System.nanoTime() + NANOSECONDS.convert(delta, MILLISECONDS);
        sequence.add(this);
    }
    @Override
    public long getDelay(@NotNull TimeUnit unit) {
        // 过期时间
        return unit.convert(trigger - System.nanoTime(), NANOSECONDS);
    }
    /**
     * 用于在过期任务队列中，任务之间执行顺序的排序
     */
    @Override
    public int compareTo(@NotNull Delayed arg) {
        DelayedTask that = (DelayedTask)arg;
        if(trigger < that.trigger) return -1;
        if(trigger > that.trigger) return 1;
        return 0;
    }
    @Override
    public void run() {
        System.out.println(this + " ");
    }
    @Override
    public String toString() {
        // 语法规则： %[argument_index$][flags][width][.precision]conversion
        // %后面的1$指的是第一个参数，也就是delta
        // $后面的-4d指的是如果数据总长度不够4位，则由右向左补足空格
        return String.format("[%1$-4d]", delta) + " Task " + id;
    }
    public String summary() {
        return "(" + id + ":" + delta + ")";
    }
    public static class EndSentinel extends DelayedTask {
        private ExecutorService exec;
        public EndSentinel(int delay, ExecutorService e) {
            super(delay);
            exec = e;
        }
        @Override
        public void run() {
            for(DelayedTask pt : sequence) {
                System.out.println(pt.summary() + " ");
            }
            System.out.println();
            System.out.println(this + " Calling shutdownNow()");
            exec.shutdownNow();
        }
    }
}

class DelayedTaskConsumer implements Runnable {
    private DelayQueue<DelayedTask> q;
    public DelayedTaskConsumer(DelayQueue<DelayedTask> q) {
        this.q = q;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                // DelayQueue take()
                // 取出队列中的head元素，若队列中没有任何延迟到期的元素存在，则该方法将会被阻塞
                // 此时有两种可能: 1. 队列中没有元素 2. 队列中有元素，但都尚未过期
                // 直到队列中有延迟到期的元素为止
                // 即便是不让DelayedTask实现Runnable接口，本例的执行结果也不会发生改变。
                // 因为此处直接调用了run()方法，并没有分配其他线程去驱动DelayedTask
                q.take().run();
            }
        } catch (InterruptedException e) {
            // Acceptable way to exit
            System.out.println("Acceptable way to exit");
        }
        System.out.println("Finished DelayedTaskConsumer");
    }
}

public class DelayQueueDemo {
    public static void main(String[] args) {
        test2();
    }
    public static void test1() {
        Random random = new Random(47);
        ExecutorService exec = Executors.newCachedThreadPool();
        DelayQueue<DelayedTask> queue = new DelayQueue<>();
        // Fill with tasks that have random delays:
        for(int i = 0; i < 5; i++) {
            queue.put(new DelayedTask(random.nextInt(5000)));
        }
        //使用DelayQueue take()的方式获取任务
        queue.add(new DelayedTask.EndSentinel(5000, exec));
        exec.execute(new DelayedTaskConsumer(queue));
    }
    public static void test2() {
        Random random = new Random(47);
        ExecutorService exec = Executors.newCachedThreadPool();
        DelayQueue<DelayedTask> queue = new DelayQueue<>();
        // Fill with tasks that have random delays:
        for(int i = 0; i < 5; i++) {
            queue.put(new DelayedTask(random.nextInt(5000)));
        }
        //使用DelayQueue poll()方式获取任务
        while(queue.size() != 0) {
            // 每执行一次DelayQueue poll()且返回的元素不是null，则DelayQueue等待队列的元素个数会减一
            DelayedTask task = queue.poll();
            if(task != null) {
                System.out.println(LocalDateTime.now(ZoneId.of("+08:00")));
            }
        }
    }
}