前言

預設的加鎖邏輯是非公平的。

在加鎖失敗時，執行緒會進入 while 迴圈，一直嘗試獲得鎖，這時候是多執行緒進行競爭。就是說誰搶到就是誰的。

Redisson 提供了 公平鎖 機制，使用方式如下：

RLock fairLock = redisson.getFairLock("anyLock");
// 最常見的使用方法
fairLock.lock();

下面一起看下公平鎖是如何實現的？

公平鎖

相信小夥伴們看過前面的文章，已經輕車熟路了，直接定位到原始碼方法：RedissonFairLock#tryLockInnerAsync。

好傢伙，這一大塊程式碼，我截圖也截不完，咱們直接分析 lua 指令碼。

PS：雖然咱不懂 lua，但是這一堆堆的 if else 咱們大概還是能看懂的。

因為 debug 發現 command == RedisCommands.EVAL_LONG，所以直接看下面一部分。

這麼長，連呼好幾聲好傢伙！

先來看看引數都有啥？

1. KEYS[1]：加鎖的名字，anyLock；
2. KEYS[2]：加鎖等待佇列，redisson_lock_queue:{anyLock}；
3. KEYS[3]：等待佇列中執行緒鎖時間的 set 集合，redisson_lock_timeout:{anyLock}，是按照鎖的時間戳存放到集合中的；
4. ARGV[1]：鎖超時時間 30000；
5. ARGV[2]：UUID:ThreadId 組合 a3da2c83-b084-425c-a70f-5d9a08b37f31:1
   ；
6. ARGV[3]：threadWaitTime 預設 300000；
7. ARGV[4]：currentTime 當前時間戳。

加鎖佇列和集合是含有大括號的字串。{XXXX} 是指這個 key 僅使用 XXXX 用來計算 slot 的位置。

Lua 指令碼分析

上面的 lua 指令碼是分為幾塊的，咱們分別從不同的角度看下上面程式碼的執行。

首次加鎖（Thread1）

第一部分，因為是首次加鎖，所以等待佇列為空，直接 跳出迴圈。這一部分執行結束。

第二部分：

1. 當鎖不存在，等待佇列為空或隊首是當前執行緒，兩個條件都滿足時，進入內部邏輯；
2. 從等待佇列和超時集合中刪除當前執行緒，這時候等待佇列和超時集合都是空的，不需要任何操作；
3. 減少佇列中所有等待執行緒的超時時間，也不需要任何操作；
4. 加鎖並設定超時時間。

執行完這裡就 return 了。所以後面幾部分就暫時不看了。

相當於下面兩個命令（整個 lua 指令碼都是原子的！）：

> hset anyLock a3da2c83-b084-425c-a70f-5d9a08b37f31:1 1
> pexpire anyLock 30000

Thread2 加鎖

當 Thread1 加鎖完成之後，此時 Thread2 來加鎖。

Thread2 可以是本例項其他執行緒，也可以是其他例項的執行緒。

第一部分，雖然鎖被 Thread1 佔用了，但是等待佇列是空的，直接跳出迴圈。

第二部分，鎖存在，直接跳過。

第三部分，執行緒是否持鎖，沒有持鎖，直接跳過。

第四部分，執行緒是否在等待佇列中，Thread2 才來加鎖，不在裡面，直接跳過。

Thread2 最後會來到這裡：

1. 從執行緒等待佇列 redisson_lock_queue:{anyLock} 中獲取最後一個執行緒；
2. 因為等待佇列是空的，所以直接獲取當前鎖的剩餘時間 ttl anyLock；
3. 組裝超時時間 timeout = ttl + 300000 + 當前時間戳，這個 300000 是預設 60000*5；
4. 使用 zadd 將 Thread2 放到等待執行緒有序集合，然後使用 rpush 將 Thread2 再放到等待佇列中。

zadd KEYS[3] timeout ARGV[2]

這裡使用 zadd 命令分別放置的是，redisson_lock_timeout:{anyLock}，超時時間戳（1624612689520），執行緒（UUID2:Thread2）。

其中超時時間戳當分數，用來在有序集合中排序，表示加鎖的順序。

Thread3 加鎖

Thread1 佔有了鎖，Thread2 在等待，此時執行緒 3 來了。

獲取 firstThreadId2 此時佇列是有執行緒的是 UUID2:Thread2。

判斷 firstThreadId2 的分數（超時時間戳）是不是小於當前時間戳：

1. 小於等於則說明超時了，移除 firstThreadId2；
2. 大於，則會進入後續判斷。

第二、三、四部分都不滿足條件。

Thread3 最後也會來到這裡：

1. 從執行緒等待佇列 redisson_lock_queue:{anyLock} 中獲取最後一個執行緒；
2. 最後一個執行緒存在，且不是自己，則 ttl = lastThreadId 超時時間戳 - 當前時間戳，就是看最後一個執行緒還有多久超時；
3. 組裝超時時間 timeout = ttl + 300000 + 當前時間戳，這個 300000 是預設 60000*5，在最後一個執行緒的超時時間上加上 300000 以及當前時間戳，就是 Thread3 的超時時間戳。
4. 使用 zadd 將 Thread3 放到等待執行緒有序集合，然後使用 rpush 將 Thread3 再放到等待佇列中。

總結

本文主要總結了公平鎖的加鎖邏輯，這涉及到比較多的 Redis 操作，做一下簡要總結：

1. Redis Hash 資料結構：存放當前鎖，Redis Key 就是鎖，Hash 的 field 是加鎖執行緒，Hash 的 value 是 重入次數；
2. Redis List 資料結構：充當執行緒等待佇列，新的等待執行緒會使用 rpush 命令放在佇列右邊；
3. Redis sorted set 有序集合資料結構：存放等待執行緒的順序，分數 score 用來是等待執行緒的超時時間戳。

需要理解的就是這裡會額外新增一個等待佇列，以及有序集合。

對照著 Java 公平鎖原始碼閱讀，理解起來效果更好。

相關推薦

• Redisson 分散式鎖原始碼 04：可重入鎖釋放
• Redisson 分散式鎖原始碼 03：可重入鎖互斥
• Redisson 分散式鎖原始碼 02：看門狗

作者： 劉志航

公眾號：『 程式設計師小航 』

個人小站：https://liuzhihang.com/

版權宣告：本部落格所有文章除特別宣告外，均採用 CC BY-NC-SA 4.0 許可協議。轉載請註明來自 Notes！