DelayQueue基本原理

DelayQueue是一個沒有邊界BlockingQueue實現(xiàn)，加入其中的元素必需實現(xiàn)Delayed接口。當生產(chǎn)者線程調(diào)用put之類的方法加入元素時，會觸發(fā)Delayed接口中的compareTo方法進行排序，也就是說隊列中元素的順序是按到期時間排序的，而非它們進入隊列的順序。排在隊列頭部的元素是最早到期的，越往后到期時間赿晚。

消費者線程查看隊列頭部的元素，注意是查看不是取出。然后調(diào)用元素的getDelay方法，如果此方法返回的值?。盎蛘叩扔冢?，則消費者線程會從隊列中取出此元素，并進行處理。如果getDelay方法返回的值大于0，則消費者線程wait返回的時間值后，再從隊列頭部取出元素，此時元素應該已經(jīng)到期。

DelayQueue是Leader-Followr模式的變種，消費者線程處于等待狀態(tài)時，總是等待最先到期的元素，而不是長時間的等待。消費者線程盡量把時間花在處理任務(wù)上，最小化空等的時間，以提高線程的利用效率。

以下通過隊列及消費者線程狀態(tài)變化大致說明一下DelayQueue的運行過程。

初始狀態(tài)

因為隊列是沒有邊界的，向隊列中添加元素的線程不會阻塞，添加操作相對簡單，所以此圖不考慮向隊列添加元素的生產(chǎn)者線程。假設(shè)現(xiàn)在共有三個消費者線程。

隊列中的元素按到期時間排序，隊列頭部的元素2s以后到期。消費者線程１查看了頭部元素以后，發(fā)現(xiàn)還需要2s才到期，于是它進入等待狀態(tài)，2s以后醒來，等待頭部元素到期的線程稱為Leader線程。

消費者線程2與消費者線程3處于待命狀態(tài)，它們不等待隊列中的非頭部元素。當消費者線程１拿到對象5以后，會向它們發(fā)送signal。這個時候兩個中的一個會結(jié)束待命狀態(tài)而進入等待狀態(tài)。

2S以后

消費者線程１已經(jīng)拿到了對象５，從等待狀態(tài)進入處理狀態(tài)，處理它取到的對象５，同時向消費者線程2與消費者線程3發(fā)送signal。

消費者線程2與消費者線程3會爭搶領(lǐng)導權(quán)，這里是消費者線程2進入等待狀態(tài)，成為Leader線程，等待2s以后對象4到期。而消費者線程3則繼續(xù)處于待命狀態(tài)。

此時隊列中加入了一個新元素對象６，它10s后到期，排在隊尾。

又2S以后

先看線程１，如果它已經(jīng)結(jié)束了對象5的處理，則進入待命狀態(tài)。如果還沒有結(jié)束，則它繼續(xù)處理對象5。

消費線程２取到對象4以后，也進入處理狀態(tài)，同時給處于待命狀態(tài)的消費線程３發(fā)送信號，消費線程３進入等待狀態(tài)，成為新的Leader?，F(xiàn)在頭部元素是新插入的對象7，因為它1s以后就過期，要早于其它所有元素，所以排到了隊列頭部。

又1S后

一種不好的結(jié)果：

消費線程３一定正在處理對象7。消費線程１與消費線程２還沒有處理完它們各自取得的對象，無法進入待命狀態(tài)，也更加進入不了等待狀態(tài)。此時對象3馬上要到期，那么如果它到期時沒有消費者線程空下來，則它的處理一定會延期。

可以想見，如果元素進入隊列的速度很快，元素之間的到期時間相對集中，而處理每個到期元素的速度又比較慢的話，則隊列會越來越大，隊列后邊的元素延期處理的時間會越來越長。

另外一種好的結(jié)果：

消費線程１與消費線程２很快的完成對取出對象的處理，及時返回重新等待隊列中的到期元素。一個處于等待狀態(tài)(Leader)，對象3一到期就立刻處理。另一個則處于待命狀態(tài)。

這樣，每一個對象都能在到期時被及時處理，不會發(fā)生明顯的延期。

所以，消費者線程的數(shù)量要夠，處理任務(wù)的速度要快。否則，隊列中的到期元素無法被及時取出并處理，造成任務(wù)延期、隊列元素堆積等情況。

示例代碼

DelayQueue的一個應用場景是定時任務(wù)調(diào)度。本例中先讓主線程向DelayQueue添加１０個任務(wù)，任務(wù)之間的啟動間隔在1~2s之間，每個任務(wù)的執(zhí)行時間固定為2s，代碼如下：  

package com.zhangdb.thread;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
class DelayTask implements Delayed {
	private static long currentTime = System.currentTimeMillis();
	protected final String taskName;
	protected final int timeCost;
	protected final long scheduleTime;
	protected static final AtomicInteger taskCount = new AtomicInteger(0);
	// 定時任務(wù)之間的啟動時間間隔在1~2s之間，timeCost表示處理此任務(wù)需要的時間，本示例中為2s
	public DelayTask(String taskName, int timeCost) {
		this.taskName = taskName;
		this.timeCost = timeCost;
		taskCount.incrementAndGet();
		currentTime += 1000 + (long) (Math.random() * 1000);
		scheduleTime = currentTime;
	}
	@Override
	public int compareTo(Delayed o) {
		return (int) (this.scheduleTime - ((DelayTask) o).scheduleTime);
	}
	@Override
	public long getDelay(TimeUnit unit) {
		long expirationTime = scheduleTime - System.currentTimeMillis();
		return unit.convert(expirationTime, TimeUnit.MILLISECONDS);
	}
	public void execTask() {
		long startTime = System.currentTimeMillis();
		System.out.println("Task " + taskName + ": schedule_start_time=" + scheduleTime + ",real start time="
				+ startTime + ",delay=" + (startTime - scheduleTime));
		try {
			Thread.sleep(timeCost);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
class DelayTaskComsumer extends Thread {
	private final BlockingQueue<DelayTask> queue;
	public DelayTaskComsumer(BlockingQueue<DelayTask> queue) {
		this.queue = queue;
	}
	@Override
	public void run() {
		DelayTask task = null;
		try {
			while (true) {
				task = queue.take();
				task.execTask();
				DelayTask.taskCount.decrementAndGet();
			}
		} catch (InterruptedException e) {
			System.out.println(getName() + " finished");
		}
	}
}
public class DelayQueueExample {
	public static void main(String[] args) {
		BlockingQueue<DelayTask> queue = new DelayQueue<DelayTask>();
		for (int i = 0; i < 10; i++) {
			try {
				queue.put(new DelayTask("work " + i, 2000));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		ThreadGroup g = new ThreadGroup("Consumers");
		for (int i = 0; i < 1; i++) {
			new Thread(g, new DelayTaskComsumer(queue)).start();
		}
		while (DelayTask.taskCount.get() > 0) {
			try {
				Thread.sleep(50);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		g.interrupt();
		System.out.println("Main thread finished");
	}
}

首先啟動一個消費者線程。因為消費者線程處單個任務(wù)的時間為2s，而任務(wù)的調(diào)度間隔為1~2s。這種情況下，每當消費者線程處理完一個任務(wù)，回頭再從隊列中新取任務(wù)時，新任務(wù)肯定延期了，無法按給定的時間調(diào)度任務(wù)。而且越往后情況越嚴重。運行代碼看一下輸出：

Task work 0: schedule_start_time=1554203579096,real start time=1554203579100,delay=4
Task work 1: schedule_start_time=1554203580931,real start time=1554203581101,delay=170
Task work 2: schedule_start_time=1554203582884,real start time=1554203583101,delay=217
Task work 3: schedule_start_time=1554203584660,real start time=1554203585101,delay=441
Task work 4: schedule_start_time=1554203586075,real start time=1554203587101,delay=1026
Task work 5: schedule_start_time=1554203587956,real start time=1554203589102,delay=1146
Task work 6: schedule_start_time=1554203589041,real start time=1554203591102,delay=2061
Task work 7: schedule_start_time=1554203590127,real start time=1554203593102,delay=2975
Task work 8: schedule_start_time=1554203591903,real start time=1554203595102,delay=3199
Task work 9: schedule_start_time=1554203593577,real start time=1554203597102,delay=3525
Main thread finished
Thread-0 finished

最后一個任務(wù)的延遲時間已經(jīng)超過3.5s了。

再作一次測試，將消費者線程的個數(shù)調(diào)整為２，這時任務(wù)應該能按時啟動，延遲應該很小，運行程序看一下結(jié)果：

Task work 0: schedule_start_time=1554204395427,real start time=1554204395430,delay=3
Task work 1: schedule_start_time=1554204396849,real start time=1554204396850,delay=1
Task work 2: schedule_start_time=1554204398050,real start time=1554204398051,delay=1
Task work 3: schedule_start_time=1554204399590,real start time=1554204399590,delay=0
Task work 4: schedule_start_time=1554204401289,real start time=1554204401289,delay=0
Task work 5: schedule_start_time=1554204402883,real start time=1554204402883,delay=0
Task work 6: schedule_start_time=1554204404663,real start time=1554204404664,delay=1
Task work 7: schedule_start_time=1554204406154,real start time=1554204406154,delay=0
Task work 8: schedule_start_time=1554204407991,real start time=1554204407991,delay=0
Task work 9: schedule_start_time=1554204409540,real start time=1554204409540,delay=0
Main thread finished
Thread-0 finished
Thread-2 finished

基本上按時啟動，最大延遲為3毫秒，大部分都是0毫秒。

將消費者線程個數(shù)調(diào)整成３個，運行看一下結(jié)果：

Task work 0: schedule_start_time=1554204499695,real start time=1554204499698,delay=3
Task work 1: schedule_start_time=1554204501375,real start time=1554204501376,delay=1
Task work 2: schedule_start_time=1554204503370,real start time=1554204503371,delay=1
Task work 3: schedule_start_time=1554204504860,real start time=1554204504861,delay=1
Task work 4: schedule_start_time=1554204506419,real start time=1554204506420,delay=1
Task work 5: schedule_start_time=1554204508191,real start time=1554204508192,delay=1
Task work 6: schedule_start_time=1554204509495,real start time=1554204509496,delay=1
Task work 7: schedule_start_time=1554204510663,real start time=1554204510664,delay=1
Task work 8: schedule_start_time=1554204512598,real start time=1554204512598,delay=0
Task work 9: schedule_start_time=1554204514276,real start time=1554204514277,delay=1
Main thread finished
Thread-0 finished
Thread-2 finished
Thread-4 finished

大部分延遲時間變成1毫秒，情況好像還不如2個線程的情況。

將消費者線程數(shù)調(diào)整成5，運行看一下結(jié)果：

Task work 0: schedule_start_time=1554204635015,real start time=1554204635019,delay=4
Task work 1: schedule_start_time=1554204636856,real start time=1554204636857,delay=1
Task work 2: schedule_start_time=1554204637968,real start time=1554204637970,delay=2
Task work 3: schedule_start_time=1554204639758,real start time=1554204639759,delay=1
Task work 4: schedule_start_time=1554204641089,real start time=1554204641090,delay=1
Task work 5: schedule_start_time=1554204642879,real start time=1554204642880,delay=1
Task work 6: schedule_start_time=1554204643941,real start time=1554204643942,delay=1
Task work 7: schedule_start_time=1554204645006,real start time=1554204645007,delay=1
Task work 8: schedule_start_time=1554204646309,real start time=1554204646310,delay=1
Task work 9: schedule_start_time=1554204647537,real start time=1554204647538,delay=1
Thread-2 finished
Thread-0 finished
Main thread finished
Thread-8 finished
Thread-4 finished
Thread-6 finished

與３個消費者線程的情況差不多。

結(jié)論

最優(yōu)的消費者線程的個數(shù)與任務(wù)啟動的時間間隔好像存在這樣的關(guān)系：單個任務(wù)處理時間的最大值　/   相鄰任務(wù)的啟動時間最小間隔?。健∽顑?yōu)線程數(shù)，如果最優(yōu)線程數(shù)是小數(shù)，則取整數(shù)后加１，比如1.3的話，那么最優(yōu)線程數(shù)應該是2。

本例中，單個任務(wù)處理時間的最大值固定為2s。 相鄰任務(wù)的啟動時間最小間隔為1s。 則消費者線程數(shù)為2/1=2。

如果消費者線程數(shù)小于此值，則來不及處理到期的任務(wù)。如果大于此值，線程太多，在調(diào)度、同步上花更多的時間，無益改善性能。

到此這篇關(guān)于Java中的DelayQueue實現(xiàn)原理及應用場景詳解的文章就介紹到這了,更多相關(guān)DelayQueue實現(xiàn)原理及應用場景內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論