常用的Redis客戶端的並發模型(轉)
偽代碼模型
# get lock lock = 0 while lock != 1: timestamp = current Unix time + lock timeout + 1 lock = SETNX lock.foo timestamp if lock == 1 or (now() > (GET lock.foo) and now() > (GETSET lock.foo timestamp)): break; else: sleep(10ms) # do your job do_job() # releaseif now() < GET lock.foo: DEL lock.foo
並發訪問
Redis為單進程單線程模式，采用隊列模式將並發訪問變為串行訪問。Redis本身沒有鎖的概念，Redis對於多個客戶端連接並不存在競爭，但是在Jedis客戶端對Redis進行並發訪問時會發生連接超時、數據轉換錯誤、阻塞、客戶端關閉連接等問題，這些問題均是由於客戶端連接混亂造成。對此有2種解決方法：
1.客戶端角度，為保證每個客戶端間正常有序與Redis進行通信，對連接進行池化，同時對客戶端讀寫Redis操作采用內部鎖synchronized。
2.服務器角度，利用setnx實現鎖。如某客戶端要獲得一個名字list的鎖，客戶端使用下面的命令進行獲取：
Setnx lock.list current time + lock timeout
如返回1，則該客戶端獲得鎖，把lock. list的鍵值設置為時間值表示該鍵已被鎖定，該客戶端最後可以通過DEL lock.list來釋放該鎖。
如返回0，表明該鎖已被其他客戶端取得，等對方完成或等待鎖超時。
第二種需要用到Redis的setnx命令，但是需要註意一些問題。
SETNX命令（SET if Not eXists）
語法：SETNX key value
功能：將 key 的值設為 value ，當且僅當 key 不存在；若給定的 key 已經存在，則 SETNX 不做任何動作。
返回值：設置成功，返回 1 。設置失敗，返回 0 。
模式：將 SETNX 用於加鎖(locking)，SETNX 可以用作加鎖原語(locking primitive)。比如說，要對關鍵字(key) foo 加鎖，客戶端可以嘗試以下方式：
SETNX lock.foo <current Unix time + lock timeout + 1>
如果 SETNX 返回 1 ，說明客戶端已經獲得了鎖， key 設置的unix時間則指定了鎖失效的時間。之後客戶端可以通過 DEL lock.foo 來釋放鎖。
如果 SETNX 返回 0 ，說明 key 已經被其他客戶端上鎖了。如果鎖是非阻塞(non blocking lock)的，我們可以選擇返回調用，或者進入一個重試循環，直到成功獲得鎖或重試超時(timeout)。
但是已經證實僅僅使用SETNX加鎖帶有競爭條件，在特定的情況下會造成錯誤。
處理死鎖(deadlock)
上面的鎖算法有一個問題：如果因為客戶端失敗、崩潰或其他原因導致沒有辦法釋放鎖的話，怎麽辦？
這種狀況可以通過檢測發現——因為上鎖的 key 保存的是 unix 時間戳，假如 key 值的時間戳小於當前的時間戳，表示鎖已經不再有效。
但是，當有多個客戶端同時檢測一個鎖是否過期並嘗試釋放它的時候，我們不能簡單粗暴地刪除死鎖的 key ，再用 SETNX 上鎖，因為這時競爭條件(race condition)已經形成了：
C1 和 C2 讀取 lock.foo 並檢查時間戳， SETNX 都返回 0 ，因為它已經被 C3 鎖上了，但 C3 在上鎖之後就崩潰(crashed)了。
C1 向 lock.foo 發送 DEL 命令。
C1 向 lock.foo 發送 SETNX 並成功。
C2 向 lock.foo 發送 DEL 命令。
C2 向 lock.foo 發送 SETNX 並成功。
出錯：因為競爭條件的關系，C1 和 C2 兩個都獲得了鎖。
幸好，以下算法可以避免以上問題。來看看我們聰明的 C4 客戶端怎麽辦：
C4 向 lock.foo 發送 SETNX 命令。
因為崩潰掉的 C3 還鎖著 lock.foo ，所以 Redis 向 C4 返回 0 。
C4 向 lock.foo 發送 GET 命令，查看 lock.foo 的鎖是否過期。如果不，則休眠(sleep)一段時間，並在之後重試。
另一方面，如果 lock.foo 內的 unix 時間戳比當前時間戳老，C4 執行以下命令：
GETSET lock.foo <current Unix timestamp + lock timeout + 1>
因為 GETSET 的作用，C4 可以檢查看 GETSET 的返回值，確定 lock.foo 之前儲存的舊值仍是那個過期時間戳，如果是的話，那麽 C4 獲得鎖。
如果其他客戶端，比如 C5，比 C4 更快地執行了 GETSET 操作並獲得鎖，那麽 C4 的 GETSET 操作返回的就是一個未過期的時間戳(C5 設置的時間戳)。C4 只好從第一步開始重試。註意，即便 C4 的 GETSET 操作對 key 進行了修改，這對未來也沒什麽影響。
這裏假設鎖key對應的value沒有實際業務意義，否則會有問題，而且其實其value也確實不應該用在業務中。
為了讓這個加鎖算法更健壯，獲得鎖的客戶端應該常常檢查過期時間以免鎖因諸如 DEL 等命令的執行而被意外解開，因為客戶端失敗的情況非常復雜，不僅僅是崩潰這麽簡單，還可能是客戶端因為某些操作被阻塞了相當長時間，緊接著 DEL 命令被嘗試執行(但這時鎖卻在另外的客戶端手上)。
GETSET命令
語法：GETSET key value
功能：將給定 key 的值設為 value ，並返回 key 的舊值(old value)。當 key 存在但不是字符串類型時，返回一個錯誤。
時間復雜度：O(1)
返回值：返回給定 key 的舊值；當 key 沒有舊值時，也即是， key 不存在時，返回 nil 。
用SETNX實現分布式鎖
Redis有一系列的命令，特點是以NX結尾，NX是Not eXists的縮寫，如SETNX命令就應該理解為：SET if Not eXists。這系列的命令非常有用，這裏講使用SETNX來實現分布式鎖。
用SETNX實現分布式鎖
利用SETNX非常簡單地實現分布式鎖。例如：某客戶端要獲得一個名字foo的鎖，客戶端使用下面的命令進行獲取：
SETNX lock.foo <current Unix time + lock timeout + 1>
- 如返回1，則該客戶端獲得鎖，把lock.foo的鍵值設置為時間值表示該鍵已被鎖定，該客戶端最後可以通過DEL lock.foo來釋放該鎖。
- 如返回0，表明該鎖已被其他客戶端取得，這時我們可以先返回或進行重試等對方完成或等待鎖超時。
解決死鎖
上面的鎖定邏輯有一個問題：如果一個持有鎖的客戶端失敗或崩潰了不能釋放鎖，該怎麽解決？我們可以通過鎖的鍵對應的時間戳來判斷這種情況是否發生了，如果當前的時間已經大於lock.foo的值，說明該鎖已失效，可以被重新使用。
發生這種情況時，可不能簡單的通過DEL來刪除鎖，然後再SETNX一次，當多個客戶端檢測到鎖超時後都會嘗試去釋放它，這裏就可能出現一個競態條件,讓我們模擬一下這個場景：
C0操作超時了，但它還持有著鎖，C1和C2讀取lock.foo檢查時間戳，先後發現超時了。
C1 發送DEL lock.foo
C1 發送SETNX lock.foo 並且成功了。
C2 發送DEL lock.foo
C2 發送SETNX lock.foo 並且成功了。
這樣一來，C1，C2都拿到了鎖！問題大了！
幸好這種問題是可以避免的，讓我們來看看C3這個客戶端是怎樣做的：
C3發送SETNX lock.foo 想要獲得鎖，由於C0還持有鎖，所以Redis返回給C3一個0
C3發送GET lock.foo 以檢查鎖是否超時了，如果沒超時，則等待或重試。
反之，如果已超時，C3通過下面的操作來嘗試獲得鎖：
GETSET lock.foo <current Unix time + lock timeout + 1>
通過GETSET，C3拿到的時間戳如果仍然是超時的，那就說明，C3如願以償拿到鎖了。
如果在C3之前，有個叫C4的客戶端比C3快一步執行了上面的操作，那麽C3拿到的時間戳是個未超時的值，這時，C3沒有如期獲得鎖，需要再次等待或重試。留意一下，盡管C3沒拿到鎖，但它改寫了C4設置的鎖的超時值，不過這一點非常微小的誤差帶來的影響可以忽略不計。
註意：為了讓分布式鎖的算法更穩鍵些，持有鎖的客戶端在解鎖之前應該再檢查一次自己的鎖是否已經超時，再去做DEL操作，因為可能客戶端因為某個耗時的操作而掛起，操作完的時候鎖因為超時已經被別人獲得，這時就不必解鎖了。
示例偽代碼
根據上面的代碼，我寫了一小段Fake代碼來描述使用分布式鎖的全過程：
# get lock
lock = 0
while lock != 1:
timestamp = current Unix time + lock timeout + 1
lock = SETNX lock.foo timestamp
if lock == 1 or (now() > (GET lock.foo) and now() > (GETSET lock.foo timestamp)):
break;
else:
sleep(10ms)
# do your job
do_job()
# release
if now() < GET lock.foo:
DEL lock.foo
是的，要想這段邏輯可以重用，使用python的你馬上就想到了Decorator，而用Java的你是不是也想到了那誰？AOP + annotation？行，怎樣舒服怎樣用吧，別重復代碼就行。
java之jedis實現
expireMsecs 鎖持有超時，防止線程在入鎖以後，無限的執行下去，讓鎖無法釋放
timeoutMsecs 鎖等待超時，防止線程饑餓，永遠沒有入鎖執行代碼的機會
/** * Acquire lock. * * @param jedis * @return true if lock is acquired, false acquire timeouted * @throws InterruptedException * in case of thread interruption */ public synchronized boolean acquire(Jedis jedis) throws InterruptedException { int timeout = timeoutMsecs; while (timeout >= 0) { long expires = System.currentTimeMillis() + expireMsecs + 1; String expiresStr = String.valueOf(expires); //鎖到期時間 if (jedis.setnx(lockKey, expiresStr) == 1) { // lock acquired locked = true; return true; } String currentValueStr = jedis.get(lockKey); //redis裏的時間 if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) { //判斷是否為空，不為空的情況下，如果被其他線程設置了值，則第二個條件判斷是過不去的 // lock is expired String oldValueStr = jedis.getSet(lockKey, expiresStr); //獲取上一個鎖到期時間，並設置現在的鎖到期時間， //只有一個線程才能獲取上一個線上的設置時間，因為jedis.getSet是同步的 if (oldValueStr != null && oldValueStr.equals(currentValueStr)) { //如過這個時候，多個線程恰好都到了這裏，但是只有一個線程的設置值和當前值相同，他才有權利獲取鎖 // lock acquired locked = true; return true; } } timeout -= 100; Thread.sleep(100); } return false; }
一般用法
其中很多繁瑣的邊緣代碼，包括：異常處理，釋放資源等等 。
JedisPool pool; JedisLock jedisLock = new JedisLock(pool.getResource(), lockKey, timeoutMsecs, expireMsecs); try { if (jedisLock.acquire()) { // 啟用鎖 //執行業務邏輯 } else { logger.info("The time wait for lock more than [{}] ms ", timeoutMsecs); } } catch (Throwable t) { // 分布式鎖異常 logger.warn(t.getMessage(), t); } finally { if (jedisLock != null) { try { jedisLock.release();// 則解鎖 } catch (Exception e) { } } if (jedis != null) { try { pool.returnResource(jedis);// 還到連接池裏 } catch (Exception e) { } } }
犀利用法
用匿名類來實現，代碼非常簡潔 至於SimpleLock的實現
SimpleLock lock = new SimpleLock(key); lock.wrap(new Runnable() { @Override public void run() { //此處代碼是鎖上的 } });
常用的Redis客戶端的並發模型(轉)