深入淺出ScheduledThreadPoolExecutor

一、ScheduledThreadPoolExecutor是什麼

ScheduledThreadPoolExecutor是一個多線程任務調度器，可以在指定的時間（延遲）後執行任務或定期執行任務。使用ScheduledThreadPoolExecutor可以幫助我們實現複雜的任務調度需求，如定時任務、延遲任務、周期性任務等。

ScheduledThreadPoolExecutor的主要屬性：

• corePoolSize: 線程池核心線程數，即同時執行的任務數。
• maximumPoolSize: 線程池最大線程數。
• keepAliveTime: 空閑線程的存活時間。
• ThreadFactory: 線程工廠，用於創建線程對象。
• RejectedExecutionHandler: 拒絕策略，用於決定當任務隊列已滿時如何處理新任務。

import java.util.concurrent.*;

public class ScheduledThreadPoolExample {
    public static void main(String[] args) {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

        Runnable task = () -> {
            System.out.println("Task executed at " + System.currentTimeMillis());
        };

        System.out.println("Scheduled to run after 2 seconds");
        executor.schedule(task, 2, TimeUnit.SECONDS);

        System.out.println("Scheduled to run after every 3 seconds, with initial delay of 1 second");
        executor.scheduleAtFixedRate(task, 1, 3, TimeUnit.SECONDS);

        System.out.println("Shutdown executor");
        executor.shutdown();
    }
}

二、ScheduledThreadPoolExecutor的使用

1、延遲任務

我們可以使用schedule()方法在指定的時間後執行任務，這個時間可以用時間和時間單位（TimeUnit）表示。

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

Runnable task = () -> {
    System.out.println("Task executed at " + System.currentTimeMillis());
};

System.out.println("Scheduled to run after 2 seconds");
executor.schedule(task, 2, TimeUnit.SECONDS);

上面的代碼中，我們創建了一個ScheduledExecutorService，然後通過schedule()方法延遲2秒執行任務。在任務執行時，我們會輸出任務執行時間的信息。

2、周期性任務

我們也可以使用scheduleAtFixedRate()方法定期執行任務，參數包括初始延遲時間、執行周期，以及時間單位。

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

Runnable task = () -> {
    System.out.println("Task executed at " + System.currentTimeMillis());
};

System.out.println("Scheduled to run after every 3 seconds, with initial delay of 1 second");
executor.scheduleAtFixedRate(task, 1, 3, TimeUnit.SECONDS);

上面的代碼中，我們執行了一個周期性任務，並且設置了初始延遲時間為1秒，執行周期為3秒。在任務執行時，我們同樣會輸出任務執行時間的信息。

3、使用線程池

ScheduledThreadPoolExecutor本質上仍然是一個線程池，我們可以像使用ThreadPoolExecutor一樣使用ScheduledThreadPoolExecutor。

首先，我們需要創建一個ScheduledExecutorService：

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);

然後，我們可以像ThreadPoolExecutor一樣提交任務：

Runnable task = () -> {
    System.out.println("Task executed at " + System.currentTimeMillis());
};

executor.submit(task);

當然，我們也可以使用schedule()或者scheduleAtFixedRate()方法提交任務，只需要把任務封裝為Runnable或者Callable接口即可。

4、異常處理

在任務執行過程中，如果任務拋出了異常，ScheduledThreadPoolExecutor將並不會打印異常堆棧信息，也不會將異常拋到調用方。因此，我們需要在任務中顯式地處理異常。

Runnable task = () -> {
    try {
        // ...
    } catch (Exception e) {
        e.printStackTrace();
    }
};

三、ScheduledThreadPoolExecutor的拓展

1、線程池監控

使用ScheduledThreadPoolExecutor可以實現線程池的監控，比如通過定期打印線程池的狀態來觀察線程池的健康狀態。

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
ScheduledFuture future = executor.scheduleAtFixedRate(() -> {
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
        System.out.println("Pool Size: " + threadPoolExecutor.getPoolSize());
        System.out.println("Active Count: " + threadPoolExecutor.getActiveCount());
        System.out.println("Completed Task Count: " + threadPoolExecutor.getCompletedTaskCount());
        System.out.println("Task Count: " + threadPoolExecutor.getTaskCount());
    }, 0, 1, TimeUnit.SECONDS);

// 在程序結束時停止監控線程池
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        future.cancel(true);
        executor.shutdown();
    }));

上面的代碼中，我們通過修改線程池的拒絕策略來打印線程池的狀態信息。在程序結束時，我們需要使用ShutdownHook來關閉線程池，確保程序結束時不會留下線程池中的未完成任務。

2、拒絕策略

當任務隊列已滿時，如果再提交新的任務，ScheduledThreadPoolExecutor默認的拒絕策略是拋出RejectedExecutionException異常。為了避免出現這種情況，我們可以自定義拒絕策略。

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (executor.isShutdown()) {
            // 如果線程池已經關閉，則直接拒絕任務
            throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + executor.toString());
        } else {
            // 否則，將任務放回任務隊列，嘗試繼續執行。
            executor.getQueue().offer(r);
        }
    }
}

ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2, new CustomRejectedExecutionHandler());

上面的代碼中，我們實現了一個自定義的拒絕策略CustomRejectedExecutionHandler。當任務隊列已滿時，我們會首先檢測線程池是否已經關閉。如果線程池已經關閉，則直接拋出拒絕執行異常，否則將任務放回任務隊列，等待線程池的新線程來執行。

3、定時任務比較工具

在一些需要進行定期任務處理的項目中，需要精確地掌握每個任務的執行情況。定時任務比較工具可以對比實際執行時間與定時時間之間的誤差，這樣就可以幫助我們調整任務的定時精度，保證任務的準確執行。

public class ScheduleTimeComparator implements Comparable {
    private final Long scheduleTime;
    private final Long actualTime;

    public ScheduleTimeComparator(Long scheduleTime, Long actualTime) {
        this.scheduleTime = scheduleTime;
        this.actualTime = actualTime;
    }

    public boolean isDelay() {
        return actualTime - scheduleTime > 0;
    }

    public Long getDelay() {
        return actualTime - scheduleTime;
    }

    @Override
    public int compareTo(ScheduleTimeComparator o) {
        return this.actualTime.compareTo(o.actualTime);
    }
}

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
List list = new ArrayList();

Runnable task = () -> {
    list.add(new ScheduleTimeComparator(System.currentTimeMillis() - 2000, System.currentTimeMillis()));
};

for (int i = 0; i < 10; i++) {
    executor.schedule(task, i * 100, TimeUnit.MILLISECONDS);
}

Thread.sleep(5000);

list.sort(Comparator.comparing(ScheduleTimeComparator::getDelay).reversed());

for (int i = 0; i < list.size(); i++) {
    ScheduleTimeComparator element = list.get(i);
    System.out.println(String.format("Index=%d, Scheduled Time=%d, Actual Time=%d, Delay=%d, Is Delay=%b", i, element.scheduleTime, element.actualTime, element.getDelay(), element.isDelay()));
}

上面的代碼中，我們通過創建一個ScheduleTimeComparator類來記錄定時時間和實際執行時間之間的誤差。當所有任務執行完成後，我們對誤差進行排序並輸出。

四、總結

本文介紹了ScheduledThreadPoolExecutor的使用方法和拓展，包括定時任務、周期性任務、線程池監控、拒絕策略、定時任務比較工具等。在實現複雜任務調度的過程中，ScheduledThreadPoolExecutor是一個非常實用的工具。