Timer

代碼實現(xiàn)

此處的代碼是《Java并發(fā)編程實戰(zhàn)》中的源碼

package edu.wyn.concurrent.chapter6;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;

public class OutofTime {

    public static void main(String[] args) throws InterruptedException {
        Timer timer = new Timer();

        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                throw new RuntimeException("test");
            }
        }, 1);
        TimeUnit.SECONDS.sleep(1);
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                throw new RuntimeException("test");
            }
        }, 1);
        TimeUnit.SECONDS.sleep(5);
    }
}

運行結(jié)果

image.png

存在問題

1、Timer在執(zhí)行所有的定時任務時只會創(chuàng)建一個線程。如果某個任務的執(zhí)行時間過長，那么將會破壞其他TimerTask的定時精確性（可能會快速連續(xù)執(zhí)行或者丟棄任務）。
2、TimerTask如果拋出了一個未檢查的異常，Timer會終止終止定時線程

ScheduledThreadPoolExecutor

使用代碼

改造原書中的代碼

package edu.wyn.concurrent.chapter6;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolDemo {

    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.schedule((Runnable) () -> {
            System.out.println("1-運行線程為：" + Thread.currentThread().getName());
            throw new RuntimeException("test");
        }, 1, TimeUnit.SECONDS);

        TimeUnit.SECONDS.sleep(1);
        executor.schedule((Runnable) () -> {
            System.out.println("2-運行線程為：" + Thread.currentThread().getName());
        }, 1, TimeUnit.SECONDS);
        TimeUnit.SECONDS.sleep(5);
        executor.shutdown();
    }
}

運行結(jié)果

image.png

即使第一個任務出現(xiàn)了異常，第二個任務還是正常開始執(zhí)行了

利用DelayQueue自定義實現(xiàn)

此處的代碼是參考材料中的，并非筆者實現(xiàn)，寫在此處主要是想自己記錄下，如有侵權(quán)請聯(lián)系筆者刪除

package edu.wyn.concurrent.chapter6;


import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

class DelayedTask implements Runnable, Delayed {

    private static int counter = 0;

    private final int id = counter++;


    //延遲時間
    private final int delta;

    //觸發(fā)時間
    private final long trigger;

    protected static List<DelayedTask> sequence = new ArrayList<>();

    public DelayedTask(int delayMilli) {
        this.delta = delayMilli;
        //轉(zhuǎn)換為nanoTime
        this.trigger = System.nanoTime() + TimeUnit.NANOSECONDS.convert(delayMilli, TimeUnit.MILLISECONDS);
        sequence.add(this);
    }

    @Override
    public void run() {
        System.out.println(this);
    }

    @Override
    public String toString() {
        return String.format("[%1$-4d]", delta) + "task=" + id;
    }

    public String summary() {
        return "(" + id + ":" + delta + ")";
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(trigger-System.nanoTime(), TimeUnit.NANOSECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        DelayedTask that = (DelayedTask) o;
        if (trigger < that.trigger) {
            return -1;
        }
        if (trigger > that.trigger) {
            return 1;
        }
        return 0;
    }

    public static class EndSentinel extends DelayedTask {
        private ExecutorService executorService;
        public EndSentinel(int delay, ExecutorService executorService) {
            super(delay);
            this.executorService = executorService;
        }

        @Override
        public void run() {
            for (DelayedTask task : sequence) {
                System.out.print(task.summary() + " ");
            }
            System.out.println();
            System.out.println(this + "調(diào)用shutdownNow");
            executorService.shutdownNow();
        }
    }
}

class DelayedTaskConsumer implements Runnable {

    private final DelayQueue<DelayedTask> queue;

    DelayedTaskConsumer(DelayQueue<DelayedTask> queue) {
        this.queue = queue;
    }


    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                queue.take().run();
            }
        } catch (InterruptedException e) {
            System.out.println("DelayedTaskConsumer消費結(jié)束");
            throw new RuntimeException(e);
        }
    }
}
public class DelayedQueueDemo {

    public static void main(String[] args) {
        Random random = new Random(400);
        ExecutorService executorService = Executors.newCachedThreadPool();
        DelayQueue<DelayedTask> queue = new DelayQueue<>();
        for (int i = 0; i < 20; i++) {
            queue.add(new DelayedTask(random.nextInt(5000)));
        }
        queue.add(new DelayedTask.EndSentinel(5000, executorService));
        executorService.execute(new DelayedTaskConsumer(queue));
    }
}

參考材料

Java并發(fā)編程實戰(zhàn)-第二部分 結(jié)構(gòu)化并發(fā)應用程序