無鎖隊列、自旋鎖隊列、互斥鎖隊列性能對比測試

介紹

無鎖隊列

先大致介紹一下無鎖隊列。無鎖隊列的根本是 CAS 函數——CompareAndSwap，即比較並交換，函數功能可以用 C++ 函數來說明：

它將 reg 的值與 oldval 的值進行對比，若相同則將 reg 賦新值。注意以上操作是原子操作。大部分語言都有提供 CAS 支持，不過函數原型可能有些微的不同，許多語言（包括 go）中 CAS 的返回值是標識是否賦值成功的 bool 值。

無鎖隊列則是以 CAS 來實現同步的一種隊列，我的具體實現這裏就不貼出來了，有點冗長，文末給出了源碼地址。這裏僅僅大致給出實現思路，網上關於無鎖隊列的資料很多，這裏就不詳細說了。

EnQueue(x) //進隊列改良版
{
    q = new record();
    q->value = x;
    q->next = NULL;
 
    p = tail;
    oldp = p
    do {
        while (p->next != NULL)
            p = p->next;
    } while( CAS(p.next, NULL, q) != TRUE); //如果沒有把結點鏈在尾上，再試
 
    CAS(tail, oldp, q); //置尾結點
}

DeQueue() //出隊列
{
    do{
        p = head;
        if (p->next == NULL){
            return ERR_EMPTY_QUEUE;
        }
    while( CAS(head, p, p->next) != TRUE );
    return p->next->value;
}

自旋鎖

自旋鎖是加鎖失敗時接着循環請求加鎖，直到成功。它的特點是不會釋放 CPU，故也沒有互斥鎖那種內核態切換操作，但缺點也很明顯，就是會一直佔用 CPU，理論上適用於臨界區小、不需要長時間加鎖的場景。這裏只貼鎖的相關代碼，隊列的實現就不貼了：

互斥鎖

這個沒什麼好說的，用的 golang 自帶的互斥鎖 sync.Mutex。

測試

下面將分 2 種場景進行測試：分別是高併發和低併發。高併發我用 4 個協程往隊列中 push 數據，4 個協程從隊列中 pop 數據（雖然不是很高，但足以區分性能，就沒測太高併發了，畢竟測一次等的太久也累）；低併發不好模擬，於是我乾脆極端點改爲無併發——先順序寫，再順序讀。

無併發

大致測試代碼結構如下（刪減了不關鍵的語句）：

t1 := time.Now()
	for i := 1; i <= dataNum; i++ {
		suc := queue.PushBack(i)
	}
	queue.Disable()

	for {
		val, enable := queue.PopFront()
		if !enable {
			break
		}
	}
	fmt.Println("用時：", time.Since(t1))

爲了方便對比，我特地還增加了不加鎖的隊列的測試結果。測試結果如下：（左側爲 dataNum 數據量）

可以看到數據量小的時候性能差別還不明顯，甚至 cas 還有少許的優勢。但數據量一大就很明顯的看出自旋鎖的效率會高一點，cas 次之。不過它們差別都不大。

高併發

這裏用 4 個生產者 4 個消費者共用一個隊列來模擬高併發。測試代碼結構如下：

func test() {
	wgr := sync.WaitGroup{}
	wgw := sync.WaitGroup{}
	t1 := time.Now()
	for i := 0; i < 4; i++ {
		wgr.Add(1)
		go reader(i*1000000, &wgr)
	}
	for i := 0; i < 4; i++ {
		wgw.Add(1)
		go writter(&wgw)
	}
	wgr.Wait()
	queue.Disable()
	wgw.Wait()
	fmt.Println("用時：", time.Since(t1))
}
func reader(startNum int, wg *sync.WaitGroup) {
	for i := 0; i < dataNum; i++ {
		suc := queue.PushBack(startNum + i)
		for !suc {
			suc = queue.PushBack(startNum + i)
		}
	}
	wg.Done()
}
func writter(wg *sync.WaitGroup) {
	for {
		r, enable := queue.PopFront()
		if enable == false {
			break
		}
		if r == defaultVal {
			continue
		}
	}
	wg.Done()
}

這種情況下就沒法測試無鎖隊列了，數據都不完整（已驗證）。測試結果如下，左側爲讀 / 寫協程數 * dataNum 數據量（下面讀 / 寫協程數爲 4 指總共開了 8 個協程）：

可以看到 cas 有巨大的性能優勢，甚至達到了 3 到 5 倍的性能差距，說明這個思路還是可行的！（先開始被 chan 打擊到了）反倒是自旋鎖的性能最差，這個倒有些出乎我的意料，按照我的理解在這種頻繁加鎖解鎖的情況下自旋鎖的性能應該更好纔對，若有知情人士望告知。

分析

爲了對這幾種鎖的性能特點有更深入的分析，這裏還補充了幾組測試，分別用了不同的協程數和數據量進行補充測試：

添加圖片註釋，不超過 140 字（可選）

可以很明顯的看到一個趨勢——隨着併發度增加，自旋鎖的性能急劇下降，由無併發時的與 cas 性能幾乎一樣到最後與 cas 將近 7 倍的效率差。而 mutex 和 cas 情況下，隨着併發度增加，性能影響並不大，下面將前面的測試數據重新組織一下方便對比：

添加圖片註釋，不超過 140 字（可選）

可以看到總數據量不變的情況下，併發協程數對 mutex 和 cas 的影響非常小，基本在波動範圍以內。相較之下自旋鎖就比較慘了。

總結

** 根據上面的結果來說的話，當實際競爭特別小的時候，可以考慮用自旋鎖；而併發大的時候，用無鎖隊列這種結構有很大潛在優勢。** 之所以說潛在的是因爲我也僅僅是簡單的實現某種結構，肯定有考慮不全的地方，我寫這個無鎖例子主要用於測試，也沒打算用於實際場景中。但是我儘量保證了同樣的代碼結構下，最大化各個鎖結構對性能的影響。總的來說，本文測試結果僅作參考，希望能有拋磚引玉的效果。

最後，再附上源碼地址：https://github.com/HandsomeRosin/lockfree

更新：

針對自旋鎖效率低下的問題我仔細想了想，應該是原子操作 cas 耗時的問題（畢竟在無併發情況下，cas 和真正不加鎖還是有很大的性能差距）。於是對自旋鎖的代碼進行了微調，減少了 CAS 的調用次數：（被註釋掉的是原本的代碼邏輯）

func (spin *spinMutex) lock() {
	// for !atomic.CompareAndSwapInt32(&spin.mutex, unlocked, locked) {}
BEGINING:
	for spin.mutex != unlocked {}
	if !atomic.CompareAndSwapInt32(&spin.mutex, unlocked, locked) {
		goto BEGINING
	}
}

事實證明，這樣做效率確實提高了約 1/4，不過還是改變不了它的大趨勢（與 cas 和 mutex 的性能差距依舊巨大），所以就沒更新前面的測試數據了。

不過這也佐證了 CAS 也是比較耗時的一個操作，平時還是不能肆意使用。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/gcardeidLEz-_NWBB-Si-Q

猜你喜歡