Java - NotificationBuffer With DelayQueue

Introduction

說到DelayQueue，我會想到拿來與ExecutorService一起使用，用以控制工作延遲多久被執行。為什麼我會把它與NotificationBuffer放在一起呢? 假設某一個Client有可能在短時間內送五筆相同資料過來，目前我們Server是會做五次例行工作；對於最後結果來說，其實我們只需要最後一次即可。除此之外，假如有多個Client，我希望能夠針對各別Client去計算這個時間區間。

Why DelayQueue?

從這兩點來說，我們需要一個buffer來儲存通知資料；然後這個buffer能夠根據各個Client第一次到達的時間，去決定何時讓Server取出並執行例行工作。以這樣的需求來說，DelayQueue是一個不錯的選擇。主要有以下幾個原因:

• Producer只要把通知資料放進buffer，它就可以回傳了。
• 繼承自BlockingQueue: 在buffer沒資料時，Consumer thread會處於block。這可以避免不必要的polling。
• 使用PriorityQueue儲存資料: 這使得通知資料被加入buffer時，delay最短的那筆將會優先被take出去。這減少我們自己寫程式計算這些東西。
• 根據delay時間去等待: 在每次資料被加入時，它會重新取得最短delay時間並做await。它使用必要的等待去取代不必要的polling檢查。

接下來我將說明要達成我目的的兩個重要物件:

• NotifiedObject: 通知資料物件。
• NotificationBuffer: 儲存通知資料物件的Buffer。

NotifiedObject

要實作自己的delayed物件，必須implements Dealyed介面，而它是extends Comparable介面，因此我們必須要實作compareTo與getDelay；除此之外，這個物件為了達到需求，會記載區別client的id與工作開始時間。可以參考底下程式碼:

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
 
public class NotifiedObject implements Delayed {
 
	private int oid;
	private long startTime;
 
	public NotifiedObject(int oid, int delayTime) {
		this.oid = oid;
		this.startTime = System.currentTimeMillis() + delayTime;
	}
 
	public int compareTo(Delayed o) {
	    return (int)(this.startTime - (((NotifiedObject)o).startTime));
	}
 
	@Override
	public long getDelay(TimeUnit unit) {
		long diff = startTime - System.currentTimeMillis();
		   return unit.convert(diff, TimeUnit.MILLISECONDS);
	}
 
	public int getHostObjId() {
		return oid;
	}
}

有幾個重點:

• startTime在建立物件時，會根據目前時間與預期delay時間去算出來；delay時間就是你希望要緩衝的時間區間。
• compareTo主要用途是給PriorityQueue決定先後順序用的，這可以讓DelayQueue取出最靠近現在時間的NotifiedObject。
• getDelay是讓DelayQueue知道要等待多久才要將結果回應給client。

NotificationBuffer

為了讓特定client的多次通知合併成同一個，因此我做了一個extends DelayQueue的NotificationBuffer。在這個buffer中，idCache用以紀錄已經在NotifiedObject的oid，用以在add時略過不需要的通知資料；在take後，也會將oid從idCache中移除，以確保後來的通知能夠被處理:

import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.DelayQueue;
 
public static class NotificationBuffer extends DelayQueue<NotifiedObject> {
	private Set<Integer> idCache = new CopyOnWriteArraySet<Integer>();
 
	@Override
	public boolean add(NotifiedObject notifyObject) {
		if(idCache.contains(notifyObject.getHostObjId())) {
			return false;
		}
		try {
			return super.add(notifyObject);
		} finally {
			idCache.add(notifyObject.getHostObjId());
		}
	}
 
	@Override
	public NotifiedObject take() throws InterruptedException {
		NotifiedObject ret = super.take();
		idCache.remove(ret.getHostObjId());
		return ret;
	}
}

因為我僅對通知者有興趣，因此我可以將多出來的事件直接省略；假如你對於每次的通知內容都有興趣，你可以在這個物件中將多個NotifiedObject給合併成一個。

Unit Test

我的單元測試主要有兩個目標，一個是確認重複的通知有被濾掉，第二個則是確認buffer是可以被重複使用兩次以上的:

@Test(timeout=10*1000)
public void testDuplicatedNotifiedObject() throws Exception {
	NotificationBuffer buffer = new NotificationBuffer();
 
	NotifiedObject object1 = new NotifiedObject(111, 2000);
	NotifiedObject object2 = new NotifiedObject(111, 2000);
	NotifiedObject object3 = new NotifiedObject(333, 2001);
 
	buffer.add(object3);
	buffer.add(object2);
	buffer.add(object1);
 
	long before = System.currentTimeMillis();
	assertEquals(object2, buffer.take());
	assertEquals(2000, System.currentTimeMillis()-before, 10);
 
	assertEquals(object3, buffer.take());
	assertEquals(2000, System.currentTimeMillis()-before, 10);
 
	// re-add & take
	NotifiedObject object4 = new NotifiedObject(333, 2001);
	buffer.add(object4);
	before = System.currentTimeMillis();
	assertEquals(object4, buffer.take());
	assertEquals(2000, System.currentTimeMillis()-before, 10);
}

@Test(timeout=20*1000)
public void testTakeDelayTime() throws Exception {
	NotificationBuffer buffer = new NotificationBuffer();
 
	NotifiedObject object1 = new NotifiedObject(111, 2000);
	NotifiedObject object2 = new NotifiedObject(222, 4000);
	NotifiedObject object3 = new NotifiedObject(333, 6000);
 
	buffer.add(object3);
	buffer.add(object2);
	buffer.add(object1);
 
	long before = System.currentTimeMillis();
	assertEquals(object1, buffer.take());
	assertEquals(2000, System.currentTimeMillis()-before, 10);
 
	assertEquals(object2, buffer.take());
	assertEquals(4000, System.currentTimeMillis()-before, 10);
 
	assertEquals(object3, buffer.take());
	assertEquals(6000, System.currentTimeMillis()-before, 10);
}

後記

針對這樣的需求，Camel的Aggregator是可以達到部分的效果，可以參考這篇文章；我會說部分效果的原因，是因為Aggregator無法根據不同client到達的時間去計算delay，它是以第一個client到達的時間為準。至於要相依於框架還是要自己造輪子，就看各位的考量了。

Reference

• Guide to DelayQueue