[分布式鎖的實作與原了解析]快速上手Spring Integration提供的可重入防死鎖的分布式鎖

快速上手Spring Integration提供的可重入防死鎖的分布式鎖

*分布式鎖，是分布式應用中不可獲缺的一個工具。
*典型的微服務架構中，在進行某些重要業務的時候，需要在整個微服務應用中對業務進行上鎖。
*除此之外，即使是簡單的單機項目，也有可能會同一個項目進行多部署，采用Apache或Nginx實作負債均衡，
在這種場景下，對互斥的業務操作也需要進行上鎖處理。
           

1、如果你之前沒有接觸過分布式鎖的概念，請移步其他文章。本篇文章不會給你講解什麼是分布式鎖，為什麼需要分布式鎖，以及如何實作分布式鎖

2、本篇文章簡單暴力的講解一套由SpringCloud項目團隊封裝出來的分布式鎖工具Spring Integration。你可以直接投入到生産環境中使用，如果你的團隊有已經實作的更好的分布式鎖，一般地的來說，是不需要再看這一個技術。但是如果你們團隊沒有能力自研分布式鎖，或者希望有一個成熟的分布式鎖能馬上投入生産使用，那這一套工具無疑是非常重要的

3、本篇文章不會帶你分析源碼，希望各位自行去翻閱源碼進行學習，鎖的API也非常簡單 （已更新分析源碼部分，往下看就有）

Spring Integration提供的分布式鎖的實作有如下4種實作方式：

• Gemfire
• JDBC
• Redis
• Zookeeper

● 一般地、在外面實作分布式鎖用的比較多的是Zookeeper和Redis。

● Spring Integration不需要你去關注它到底是基于什麼存儲技術實作的，它是面向接口程式設計，低耦合讓你不需要關注底層實作。你要做的僅僅是做簡單的選擇，然後用相同的一套api即可完成分布式鎖的操作。

該分布式鎖的優缺點：

1、已實作可重入、解決了死鎖問題

可重入：同一個線程，可以多次獲得相同的鎖。這個應該是實作鎖都應該去實作的特性。否則你的鎖，很有可能自己把自己搞死了。

死鎖問題：如果一個線程在競争鎖成功後，意外當機了，導緻沒有主動去釋放鎖。那麼鎖在一般情況下，就會永久保留，這就造成了死鎖。需要人工去處理，一般的，類似于使用Redis作為實作工具的，出現死鎖的時候，就要手動去Redis裡面找到這一個鎖然後del 掉它。

2、缺點，無法續期鎖

為了解決死鎖問題，在redis作為實作工具的情況下，預設是采用redis的TTL設定過期事件來解決死鎖問題。預設是60s，如果你加鎖之後的業務操作，大于60秒，就會導緻鎖自動釋放，其他線程此時可以競争獲得你的鎖。但是實際上，你本應該還持有鎖。

該架構沒有鎖續期，或者自定義鎖過期時間的API，是以要非常注意你加鎖的業務功能，務必要在60s内完成。

一般地、在其他大牛實作分布式鎖時，會有另外一個線程持續監控獲得鎖的線程，如果線程沒有主動釋放鎖，而又處于活躍狀态（即還在處理業務），那麼另外一個線程會幫助這個鎖進行續期，以保證鎖不會因為逾時而自動釋放。

（本人沒有過度研究這套源碼，可能是已經實作了續期的功能，但是我不知道在哪裡使用，如果有人知道，可以在留言區提醒我）

=========================================================================================================

啥也不說，直接開幹。

項目基于Maven+SpringBoot ， 分布式鎖的實作采用的是Redis 。（是以請為SpringBoot整合好Redis）

Step 1: 導入Spring Integration依賴

Step 2: 配置JavaConfig以及Bean

Step 3：擷取鎖的代碼骨架

在需要使用鎖的Bean裡面 注入依賴

官方源碼位置： https://github.com/spring-projects/spring-integration

Good Lucky!

上面的文章快速入門了基于Redis實作的分布式可重入鎖，你已經可以直接在生産環境中使用該鎖

以下為原理刨析，可以讓你對Spring Integration實作的分布式鎖有更深入的了解。

在閱讀了它的源碼後，本人覺得有一些很值得學習的思想

（這裡假設你已經有ReentrantLock的知識概念，以及操作Redis的知識概念，否則以下内容你将無法展開）

STEP 1：首先從RedisLockRegistry這個類出發分析

   從上文圖檔可知，RedisLockRegistry是通過new RedisLockRegistry(redisConnectionFactory, “redis-lock-test”)；

傳入的參數包括：

• RedisConnectionFactory用于構造RedisTemplate（用于操作Redis指令)
• registryKey: 你的分布式鎖在Redis中的字首，請為你的分布式應用合理的指定一個唯一的名稱字首
• expireAfter（過期時間毫秒數）：預設是DEFAULT_EXPIRE_AFTER（60秒），也就是說，你的鎖最多持有60s，如果你的正常業務代碼持鎖期間有可能會超過60s，那麼你必須使用第二個構造方法，來為你的應用指定合理的過期時間。這也是分布式鎖防止線程意外當機，出現死鎖的情況
• obtainLockScript (Lua語言實作的腳本代碼)：簡單的說，Redis的Lua本身實作了原子操作，這段腳本的功能就是實作從redis競争鎖的過程，待會會給你詳細解讀腳本代碼。此外，這個屬性是final，由官方定義好腳本語言，一般情況下，你是不需要改動的
• final String clientId = UUID.randomUUID().toString() ： 還有一個屬性定義在RedisLockRegistry對象裡，它的作用是通過UUID随機生成一個不重複的id，以此來區分不同應用程式。（由于在Spring的整合中，RedisLockRegistry是單例的，是以這裡對于每個應用程式來說，它隻有唯一的一個執行個體，是以clientId的作用就是區分不同應用程式。倘若你的應用程式執行個體化了多個RedisLockRegistry，那麼clientId的作用僅僅用于辨別不同的執行個體對象，它們的核心作用在于在Redis端的競争）

STEP 2：redisLockRegistry.obtain(String lockKey)

擷取鎖的第一步，通過lockKey定義一個即将要去競争的鎖

每個redisLockRegistry對象内部會維護一個線程安全的Map，即上面代碼中第三行的locks。它的作用是用于儲存名為lockKey對應的RedisLock對象。

可能會有人問computeIfAbsent以及RedisLock::new是什麼來的

• JDK8中Map有一個新方法computeIfAbsent，用于如果傳入的key對應的value為null，就将第二個參數設定進去
• JDK8中有個新的文法糖，專業屬于叫引用方法，可以自行百度學習。RedisLock:;new的意思就是傳入RedisLock的構造方法進去，它有一個新的類叫Function，請自行了解

通過上面代碼，你就已經獲得了一個名為lockKey的Lock對象。下面進入重點環節

STEP 3: RedisLock implements Lock詳解

• Lock接口是并發程式設計包JUC中比較常見的一個接口，很多實作鎖功能的類都是實作這個接口，它為Java程式設計裡面的鎖提供了一個抽象
• RedisLock則是Spring Integration作者根據實際項目需求所實作的鎖，它的目的就是實作分布式鎖的功能

RedisLock的3個主要屬性:

1. private final String lockKey：（全鎖名）它是完整的鎖名，它會組合你在RedisLockRegistry對象定義的registryKey（字首）+ 你obtain（）時傳入的lockKey。 是以它是完整的在Redis中的key值
2. private final ReentrantLock localLock = new ReentrantLock(); (實作可重入的核心)可重入鎖，這裡不多說。它在這裡的目的是為了實作目前用戶端的資源競争。Spring Integration實作的分布式鎖分為兩個步驟，首先線程是在目前用戶端進行競争鎖資源，競争成功後再代表目前用戶端去Redis端與其他用戶端進行鎖競争。
3. private volatile long lockedAt; (競争鎖成功那一刻的時間) 用于記錄目前鎖競争成功那一刻的時間毫秒數

lock(), tryLock()詳解：

首先無論是lock還是trylock方法，他們隻有無限阻塞和嘗試一段時間競争鎖的差別。他們的工作核心流程都是：先競争ReentrantLock，成功後再調用obtainLock（）進行Redis端的鎖競争。 兩步依次都成功後，才會傳回true，表明你本次競争鎖成功。

代碼(僅解讀lock()方法，trylock自行舉一反三):

@Override
public void lock() {
	// 第一步，先進行ReentrantLock競争，它的目的是在目前用戶端中，不同線程之間先競争一輪，決出最終競争成功的那個線程
	// 同時這ReentrantLock預設是nonFair非公平鎖
	this.localLock.lock();
	// 運作到這裡表明，目前線程已經在本用戶端中競争成功，但并不意味着，你的分布式鎖就能成功
	// 此時，你将代表目前用戶端clientId，去Redis端進行競争
	while (true) {
		try {
			// 調用obtainLock（）去redis端進行競争，直到競争成功
			while (!obtainLock()) {
				Thread.sleep(100); // 本次競争失敗，等待100ms再去嘗試
			}
			// 執行到這裡，表明Redis端也競争成功了，此刻，你才是真正的分布式鎖競争成功！
			break;
		}
		catch (InterruptedException e) {
			
		}
		catch (Exception e) {
			// 注意如果在這裡try catch出現任何異常，我們都需要把目前用戶端的ReentrantLock進行unlock釋放,防止死鎖
			this.localLock.unlock();
			rethrowAsLockException(e);
		}
	}
}

/**
* 線程在自己的用戶端中競争成功後，代表目前用戶端去Redis端進行分布式鎖的競争
*/
private boolean obtainLock() {
	// 操作redis-lua的方法
	Boolean success =
			RedisLockRegistry.this.redisTemplate.execute(
					/* lua腳本 */RedisLockRegistry.this.obtainLockScript,
					/* keys */Collections.singletonList(this.lockKey), 
					/* argv[1] */RedisLockRegistry.this.clientId,
					/* argv[2] */String.valueOf(RedisLockRegistry.this.expireAfter));

	/*
	   lua腳本執行後，會傳回true or false
	   true：分布式鎖競争成功
	   false：分布式鎖競争失敗
	*/
	boolean result = Boolean.TRUE.equals(success);

	if (result) {
		// 如果true，那麼就記錄此刻的時間
		this.lockedAt = System.currentTimeMillis();
	}
	return result;
}
           

LUA腳本代碼（已為你注釋好，應該自行閱讀問題不大）

/**
* KEYS[1] : lockKey 鎖名
* ARGV[1] : clientId 用戶端id
* ARGV[2] : expireAfter 過期時間毫秒級
*/

// 調用Redis GET指令，擷取lockKey對應的value值
local lockClientId = redis.call('GET', KEYS[1])		


if lockClientId == ARGV[1] then	
  // 如果lockClientId不為空，且value值等于傳入的clientId，說明它就是這個鎖的鎖主
  // 出現這種情況，說明這是第N(N>1)次重入
  // 在Redis端會為這次重入，重置KV的TTL
  redis.call('PEXPIRE', KEYS[1], ARGV[2])		// 調用PEXPIRE為 lockkey設定過期時間（毫秒）
  return true

elseif not lockClientId then		
  // 如果lockClientId為空，則會進入目前elseif
  // 出現這種情況，說明沒有其他用戶端持有該鎖，是以該value才會為nil（空）
  // 表明你這次競争鎖成功
  redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
  return true
end

// 如果lockClientId不為空，又不等于目前clientId
// 那麼就是競争失敗
return false
           

從上面lock（）方法的源碼可知，解決可重入問題是通過ReentrantLock來輔助實作的。而解決死鎖問題則是通過Redis的TTL實作。它的工作思想比較巧妙，總結為以下一張圖：

在這裡，簡單的說：

   假設有三所學校A,B,C。每所學校有3個教師A1,A2,A3,B1,B2,B3,C1,C2,C3。 一共9個老師去教育局請教育局長來學校調研。

   在這裡，教育局局長就是共享資源，它每次肯定隻能去一所學校參觀

   首先，每所學校的3名教師會先進行内部競争，決出一名教師代表自己的學校去教育局。最終每所學校的那名教師代表自己的學校，到達教育局，教育局局長的接待工作由秘書負責，秘書按照先後順序接見A,B,C三所學校的代表教師。如果教育局局長有空，則由最先到的教師帶走教育局局長去它的學校調研。調研結束後，教育局局長傳回教育局，再由第二所學校的教師帶走教育局局長。

我想這樣解釋就很生動的模拟了上面分布式鎖的競争過程。

⭐思考：為什麼不能是9個教師直接到教育局進行先後競争呢？

回答：

1. 開銷：每個學校派出3名老師，他們的路費就是3倍。
2. 資源：教育局的接待數量有限，而且肯定不止一種業務（邀請教育局局長到學校調研）。9個教師同時到達教育局，教育局的接客空間是有限的，而且人多起來，可能秘書會手忙腳亂。

⭐回歸整體的思考：為什麼不能是9個線程，直接到Redis端進行競争呢？

個人分析：

    首先，如果你有了解過Redis實作的分布式鎖，你可以從百度上看到很多别人的文章。最簡單也挺有效的一種方式，就是利用Redis的setnx指令以及Redis本身單線程串行處理所有指令的特性，來實作一個可用的分布式鎖。他們這些鎖都有一個特點，就是每個線程為一個個體，到達Redis進行競争。

    之是以這裡，作者要這樣設計，我想應該出于以下幾點優化：

    1. 網絡開銷：同一個應用程式3個線程，就需要發送3條指令到Redis，并且有其中2條指令是肯定會失敗的

    2. 線程自旋開銷：如果競争失敗，像lock的邏輯就是不斷去重試直到成功，那麼每次重試都需要發送一次Redis指令，每次都是網絡開銷。但是作者現在是，先是内部JVM層面的競争，競争成功後就會由3個線程變為1個線程去進行會消耗網絡的自旋。而另外2個線程則隻是消耗CPU的自旋。倘若是3個線程都去redis進行競争，那麼就是3個CPU自旋+3個網絡消耗。而現在隻是3個CPU自旋+1個網絡消耗

    3. Redis性能：Redis肯定不僅僅是為了解決分布式鎖而存在的，它的功能有很多。9個線程去讓redis進行工作，和3個線程去讓redis進行工作，對redis的性能消耗肯定是不同的。（當然這裡3和9肯定可以忽略不計了，但是畢竟這裡簡單舉例子，放大十倍，百倍，千倍就是一筆大的開銷）

    4.為了實作可重入：我想這個才是這項設計比較核心的考慮。對比網上沒有可重入功能的redis分布式鎖，可以看到都是沒有ReentrantLock的輔助的。但是我們可知，可重入性幾乎是鎖必備的特性，而ReentrantLock是Java實作好的一款極具生産價值的可重入鎖。是以作者為了利用ReentrantLock實作可重入性，而由此衍生出這樣的設計考慮。

（當然上面都是個人分析罷了，實作可重入應該還有很多方式，不過在看了作者的源碼後，感覺這是一個非常不錯的考慮）
           

unlock()方法：

@Override
public void unlock() {
	if (!this.localLock.isHeldByCurrentThread()) {
		// 先判斷目前ReentrantLock的鎖主是不是目前線程
		throw new IllegalStateException("You do not own lock at " + this.lockKey);
	}
	if (this.localLock.getHoldCount() > 1) {
		// 判斷可重入标記，如果大于1，說明重入了getholdCount()次，這一次unlock()隻是讓計數-1，而不會真正釋放Redis端的分布式鎖
		this.localLock.unlock();
		return;
	}
	try {
		if (!isAcquiredInThisProcess()) {
			// 這個方法封裝了redis端的判斷，它會判斷Redis端的鎖是不是你持有的
			// 一般情況下，這個方法都會傳回true，則跳過這次報錯

			// 如果代碼進入此報錯，原因主要是，每個鎖的過期時間預設60s，如果你持有鎖的情況下超過60s後再unlock()，
			// 此時鎖早就已經過期丢棄，甚至被其他線程競争掉，是以你的unlcok會失敗
			throw new IllegalStateException("Lock was released in the store due to expiration. " +
					"The integrity of data protected by this lock may have been compromised.");
		}

		// 調用redis del指令删除KV。也就是釋放這次分布式鎖
		if (Thread.currentThread().isInterrupted()) {
			RedisLockRegistry.this.executor.execute(() ->
					RedisLockRegistry.this.redisTemplate.delete(this.lockKey));
		}
		else {
			RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
		}

		if (logger.isDebugEnabled()) {
			logger.debug("Released lock; " + this);
		}
	}
	catch (Exception e) {
		ReflectionUtils.rethrowRuntimeException(e);
	}
	finally {
		// 最後記得把本地的ReentrantLock進行unlock(),以讓其他等待線程進行競争
		this.localLock.unlock();
	}
}