這里提供兩種在指定時間后啟動線程的方法。一是通過java.util.concurrent.DelayQueue實現;二是通過java.util.concurrent.ScheduledThreadPoolExecutor實現。
1. java.util.concurrent.DelayQueue
類DelayQueue是一個無界阻塞隊列,只有在延遲期滿時才能從中提取元素。它接受實現Delayed接口的實例作為元素。
<<interface>>Delayed.java
package java.util.concurrent; import java.util.*; public interface Delayed extends Comparable<Delayed> { long getDelay(TimeUnit unit); }
getDelay()返回與此對象相關的剩余延遲時間,以給定的時間單位表示。此接口的實現必須定義一個 compareTo 方法,該方法提供與此接口的 getDelay 方法一致的排序。
DelayQueue隊列的頭部是延遲期滿后保存時間最長的 Delayed 元素。當一個元素的getDelay(TimeUnit.NANOSECONDS) 方法返回一個小於等於 0 的值時,將發生到期。
2.設計帶有時間延遲特性的隊列
類DelayedTasker維護一個DelayQueue<DelayedTask> queue,其中DelayedTask實現了Delayed接口,並由一個內部類定義。外部類和內部類都實現Runnable接口,對於外部類來說,它的run方法是按定義的時間先后取出隊列中的任務,而這些任務即內部類的實例,內部類的run方法定義每個線程具體邏輯。
這個設計的實質是定義了一個具有時間特性的線程任務列表,而且該列表可以是任意長度的。每次添加任務時指定啟動時間即可。
DelayedTasker.java
package com.zj.timedtask; import static java.util.concurrent.TimeUnit.SECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; import java.util.Collection; import java.util.Collections; 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 DelayedTasker implements Runnable { DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>(); public void addTask(DelayedTask e) { queue.put(e); } public void removeTask() { queue.poll(); } public Collection<DelayedTask> getAllTasks() { return Collections.unmodifiableCollection(queue); } public int getTaskQuantity() { return queue.size(); } public void run() { while (!queue.isEmpty()) try { queue.take().run(); } catch (InterruptedException e) { System.out.println("Interrupted"); } System.out.println("Finished DelayedTask"); } public static class DelayedTask implements Delayed, Runnable { private static int counter = 0; private final int id = counter++; private final int delta; private final long trigger; public DelayedTask(int delayInSeconds) { delta = delayInSeconds; trigger = System.nanoTime() + NANOSECONDS.convert(delta, SECONDS); } public long getDelay(TimeUnit unit) { return unit.convert(trigger - System.nanoTime(), NANOSECONDS); } public int compareTo(Delayed arg) { DelayedTask that = (DelayedTask) arg; if (trigger < that.trigger) return -1; if (trigger > that.trigger) return 1; return 0; } public void run() { //run all that you want to do System.out.println(this); } public String toString() { return "[" + delta + "s]" + "Task" + id; } } public static void main(String[] args) { Random rand = new Random(); ExecutorService exec = Executors.newCachedThreadPool(); DelayedTasker tasker = new DelayedTasker(); for (int i = 0; i < 10; i++) tasker.addTask(new DelayedTask(rand.nextInt(5))); exec.execute(tasker); exec.shutdown(); } }
結果:
[0s]Task 1
[0s]Task 2
[0s]Task 3
[1s]Task 6
[2s]Task 5
[3s]Task 8
[4s]Task 0
[4s]Task 4
[4s]Task 7
[4s]Task 9
Finished DelayedTask
3. java.util.concurrent.ScheduledThreadPoolExecutor
該類可以另行安排在給定的延遲后運行任務(線程),或者定期(重復)執行任務。在構造子中需要知道線程池的大小。最主要的方法是:
[1] schedule
創建並執行在給定延遲后啟用的一次性操作。
指定者:
-接口 ScheduledExecutorService 中的 schedule;
參數:
-command - 要執行的任務 ;
-delay - 從現在開始延遲執行的時間 ;
-unit - 延遲參數的時間單位 ;
返回:
-表示掛起任務完成的 ScheduledFuture,並且其 get() 方法在完成后將返回 null。
[2] scheduleAtFixedRate
public ScheduledFuture<?> scheduleAtFixedRate( Runnable command,long initialDelay,long period,TimeUnit unit)
創建並執行一個在給定初始延遲后首次啟用的定期操作,后續操作具有給定的周期;也就是將在 initialDelay 后開始執行,然后在 initialDelay+period 后執行,接着在 initialDelay + 2 * period 后執行,依此類推。如果任務的任何一個執行遇到異常,則后續執行都會被取消。否則,只能通過執行程序的取消或終止方法來終止該任務。如果此任務的任何一個執行要花費比其周期更長的時間,則將推遲后續執行,但不會同時執行。
指定者:
-接口 ScheduledExecutorService 中的 scheduleAtFixedRate;
參數:
-command - 要執行的任務 ;
-initialDelay - 首次執行的延遲時間 ;
-period - 連續執行之間的周期 ;
-unit - initialDelay 和 period 參數的時間單位 ;
返回:
-表示掛起任務完成的 ScheduledFuture,並且其 get() 方法在取消后將拋出異常。
4.設計帶有時間延遲特性的線程執行者
類ScheduleTasked關聯一個ScheduledThreadPoolExcutor,可以指定線程池的大小。通過schedule方法知道線程及延遲的時間,通過shutdown方法關閉線程池。對於具體任務(線程)的邏輯具有一定的靈活性(相比前一中設計,前一種設計必須事先定義線程的邏輯,但可以通過繼承或裝飾修改線程具體邏輯設計)。
ScheduleTasker.java
package com.zj.timedtask; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class ScheduleTasker { private int corePoolSize = 10; ScheduledThreadPoolExecutor scheduler; public ScheduleTasker() { scheduler = new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduleTasker(int quantity) { corePoolSize = quantity; scheduler = new ScheduledThreadPoolExecutor(corePoolSize); } public void schedule(Runnable event, long delay) { scheduler.schedule(event, delay, TimeUnit.SECONDS); } public void shutdown() { scheduler.shutdown(); } public static void main(String[] args) { ScheduleTasker tasker = new ScheduleTasker(); tasker.schedule(new Runnable() { public void run() { System.out.println("[1s]Task 1"); } }, 1); tasker.schedule(new Runnable() { public void run() { System.out.println("[2s]Task 2"); } }, 2); tasker.schedule(new Runnable() { public void run() { System.out.println("[4s]Task 3"); } }, 4); tasker.schedule(new Runnable() { public void run() { System.out.println("[10s]Task 4"); } }, 10); tasker.shutdown(); } }
結果:
[1s]Task 1
[2s]Task 2
[4s]Task 3
[10s]Task 4
本文出自 “子 孑” 博客,請務必保留此出處http://zhangjunhd.blog.51cto.com/113473/74646