跳到主要內容

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

留言

這個網誌中的熱門文章

PostgreSQL - Unattended installation on windows

Introduction 要將別人軟體包裝到自己軟體中,不可或缺的東西就是Unattended installation。以Unattended installation來說,我們可以選擇透過Installer的silent mode安裝,也可以透過把目標軟體做成portable的版本。本篇文章分享這兩種方法,教導大家如何將PostgreSQL透過Unattended installation方式安裝到目標系統成為service。 Note. 本篇以PostgreSQL 10.7為例。 Install with installer Tips 安裝程式或反安裝程式的參數,除了可以直接上官網搜尋Installation User Guide以外,也可以直接使用help參數查詢: postgresql- 10.7 - 2 -windows-x64.exe --help Windows安裝程式主要有EnterpriseDB與BigSQL兩種。BigSQL版本安裝元件是透過網路下載且支援參數不如EnterpriseDB版本多,以我們需求來說,我們傾向於使用EnterpriseDB版本。接下來分享給大家安裝與反安裝方法。 Installation @ echo off set INSTALL_DIR =C:\postgres10 set INSTALLER =postgresql- 10.7 - 2 -windows-x64.exe   rem options for installation set SSMDB_SERVICE =postgresql- 10 set MODE =--unattendedmodeui none --mode unattended   set DB_PASSWD =--superpassword postgres set DB_PORT =--serverport 5432   set SERVICE_NAME =--servicename % SSMDB_SERVICE %   set PREFIX =--prefix "%INSTALL_DIR%" set DATA_DIR =--datadir "%INSTALL_DIR%\data"   set OPTIONS =

How to install RIDE on Windows?

Introduction 多年沒在Windows上開發RobotFramework,趁著這次整理一下RIDE安裝方法。 目前RIDE最新版本與Python對應版本如下: (3.6 < python <= 3.11) Install current released version (2.0.8.1) with: pip install -U robotframework-ride 安裝Python 直接到Python官網找尋最新的3.11版本,我使用3.11.9: link 。安裝就是一直下一步而已。 安裝wxPython 每次安裝RIDE最困難的都是wxPython。看了一下 官網 描述,我就姑且相信一下: 接著進入下 載頁面 就有安裝教學。基本上就是到Python目錄下的Scripts直接執行以下command: pip install -U wxPython 安裝RIDE 接著就如RIDE官網所說,執行以下command: pip install -U robotframework-ride 啟動RIDE 直接在相同目錄下執行ride就可以啟動了,你也可以直接在桌面建ride連結,加快下次啟動時間。 沒想到這次這麼順利就安裝完成了。因為我是使用java去啟動robot framework,就不特別講要怎麼使用pip安裝robot framework了。

Hello World!

即將要搬家,因此舊網頁內容將慢慢轉移至Blogger。 如果要存取舊網頁,可以使用以下連結: https://wiki.tonylin.idv.tw/dokuwiki/doku.php