1. 為什么需要分布式鎖？

在分布式環境下，數據一致性問題一直是個難點。分布式與單機環境最大的不同在于它不是多線程而是多進程。由于多線程可以共享堆內存，因此可以簡單地采取內存作為標記存儲位置。而多進程可能都不在同一臺物理機上，就需要將標記存儲在一個所有進程都能看到的地方。

例如秒殺場景就是一個常見的多進程場景。訂單服務部署了多個服務實例，如秒殺商品有 4 個，第一個用戶購買 3 個，第二個用戶購買 2 個，理想狀態下第一個用戶能購買成功，第二個用戶提示購買失敗，反之亦可。而實際可能出現的情況是，兩個用戶都得到庫存為 4，第一個用戶買到了 3 個，更新庫存之前，第二個用戶下了 2 個商品的訂單，更新庫存為 2，導致業務邏輯出錯。

在上面的場景中，商品的庫存是共享變量，面對高并發情況，需要保證對資源的訪問互斥。

在單機環境中，比如 Java 語言中其實提供了很多并發處理相關的 API ，但是這些 API 在分布式場景中就無能為力了。

由于分布式系統具備多線程和多進程的特點，且分布在不同機器中，synchronized 和 lock 關鍵字將失去原有鎖的效果，僅依賴這些語言自身提供的 API 并不能實現分布式鎖的功能，因此需要我們找到其他方法實現分布式鎖。

常見的鎖方案如下：

• 基于數據庫實現分布式鎖；
• 基于 ZooKeeper 實現分布式鎖；
• 基于緩存實現分布式鎖，如 redis、etcd 等；

2. 基于數據庫實現分布式鎖

基于數據庫實現分布式鎖有兩種方式:

• 基于數據庫表；
• 基于數據庫的排他鎖；

2.1 基于數據庫表的增刪

基于數據庫表的增刪是最簡單的實現方式，首先創建一張鎖的表，主要包含方法名、時間戳等字段。

具體使用的方法為：當需要鎖住某個方法時，往該表中插入一條相關的記錄。需要注意的是，方法名有唯一性約束。如果有多個請求同時提交到數據庫，數據庫會保證只有一個操作可以成功，那么我們就可以認為操作成功的那個線程獲得了該方法的鎖，可以執行業務邏輯。執行完畢，需要刪除該記錄。

對于上述方案我們可以進行優化，如應用主從數據庫，數據之間雙向同步。一旦主庫掛掉，將應用服務快速切換到從庫上。除此之外還可以記錄當前獲得鎖的機器的主機信息和線程信息，下次再獲取鎖的時候先查詢數據庫，如果當前機器的主機信息和線程信息在數據庫可以查到，直接把鎖分配給該線程，實現可重入鎖。

2.2 基于數據庫排他鎖

還可以通過數據庫的排他鎖來實現分布式鎖。基于 MySQL 的 InnoDB 引擎，可以使用以下方法來實現加鎖操作：

public void lock(){connection.setAutoCommit(false)int count = 0;while(count < 4){try{select * from lock where lock_name=xxx for update;if(結果不為空){// 代表獲取到鎖return;}}catch(Exception e){}// 為空或者拋異常都表示沒有獲取到鎖sleep(1000);count++;}throw new LockException();
}

在查詢語句后面增加 for update ，數據庫會在查詢過程中給數據庫表增加排他鎖。當某條記錄被加上排他鎖之后，其他線程就無法再在該行記錄上增加排他鎖。其他沒有獲取到鎖的線程就會阻塞在上述 select 語句上，可能出現兩種結果：在超時之前獲取到了鎖，在超時之前仍未獲取到鎖。

獲得排他鎖的線程即可獲得分布式鎖，獲取到鎖之后，可以執行業務邏輯，執行業務后釋放鎖即可。

2.3 數據庫分布式鎖總結

上面兩種方式的實現都是依賴數據庫的一張表，一種是通過表中記錄的存在情況確定當前是否有鎖存在，另外一種是通過數據庫的排他鎖來實現分布式鎖。

• 優點：直接借助現有的關系型數據庫，簡單且容易理解；
• 缺點：操作數據庫需要一定的開銷，性能問題以及 SQL 執行超時的異常需要考慮；

3. 基于 ZooKeeper 實現分布式鎖

基于 ZooKeeper 的臨時節點和順序特性可以實現分布式鎖。

申請對某個方法加鎖時，在 ZooKeeper 上與該方法對應的指定節點的目錄下，生成一個唯一的臨時有序節點。當需要獲取鎖時，只需要判斷有序節點中該節點是否為序號最小的一個。業務邏輯執行完成釋放鎖，只需將這個臨時節點刪除。這種方式也可以避免由于服務宕機導致的鎖無法釋放，產生的死鎖問題。

Netflix 開源了一套 ZooKeeper 客戶端框架 Curator，Curator 提供的 InterProcessMutex 是分布式鎖的一種實現。acquire 方法獲取鎖，release 方法釋放鎖。另外，鎖釋放、阻塞鎖、可重入鎖等問題都可以有效解決。

關于阻塞鎖的實現，客戶端可以通過在 ZooKeeper 中創建順序節點，并且在節點上綁定監聽器 Watch。一旦節點發生變化，ZooKeeper 會通知客戶端，客戶端可以檢查自己創建的節點是否是當前所有節點中序號最小的，如果是就獲取到鎖，執行業務邏輯。

ZooKeeper 實現的分布式鎖也存在一些缺陷，比如在性能上可能不如基于緩存實現的分布式鎖。因為每次創建鎖和釋放鎖的過程中，都要動態創建、銷毀瞬時節點，實現鎖功能。

此外，ZooKeeper 中創建和刪除節點只能通過 Leader 節點來執行，然后將數據同步到集群中的其他節點。分布式環境中難免存在網絡抖動，導致客戶端和 ZooKeeper 集群之間的 session 連接中斷，此時 ZooKeeper 服務端以為客戶端掛了，就會刪除臨時節點。這時其他客戶端就可以獲取到分布式鎖了，會出現多個請求獲取到了同一把鎖的問題，導致業務數據不一致。

4. 基于緩存實現分布式鎖

相對于基于數據庫實現分布式鎖的方案來說，基于緩存來實現在性能方面會表現得更好一點，存取速度會快很多，而且很多緩存是可以集群部署的，可以解決單點問題。

基于緩存的鎖有如下幾種： memcached、redis、etcd。下面我們主要講解基于 etcd 實現的分布式鎖。

通過 etcd 實現分布式鎖，同樣需要滿足一致性、互斥性和可靠性等要求。etcd 中的事務 txn、lease 租約以及 watch 監聽特性，能夠實現上述要求的分布式鎖。

4.1 整體思路

通過 etcd 的事務特性可以幫助我們實現一致性和互斥性。etcd 的事務特性，使用 IF-Then-Else 語句，IF 語言判斷 etcd 服務端是否存在指定的 key，通過該 key 創建的版本號 create_revision 是否為 0 來檢查 key 是否已存在，如果該 key 存在，版本號不為 0。滿足 IF 條件的情況下則使用 Then 執行 put 操作，否則 Else 語句將返回搶鎖失敗的結果。

當然，除了使用 key 是否創建成功作為 IF 的判斷依據，還可以創建前綴相同的 key，通過比較這些 key 的 revision 來判斷分布式鎖應該屬于哪個請求。

客戶端請求在獲取到分布式鎖后，如果發生異常，需要及時將鎖釋放掉，因此需要租約。我們申請分布式鎖時也需要指定租約時間，超過 lease 租期時間將會自動釋放鎖，保證業務的可用性。

但是在執行業務邏輯時，如果客戶端發起的是一個耗時的操作，在操作未完成的情況下，租約時間過期，就會導致其他請求獲取到分布式鎖，造成不一致。這種情況下就需要續租，即刷新租約，使得客戶端和 etcd 服務端持續保持心跳。

4.2 代碼實現

package mainimport ("context""fmt""time""github.com/coreos/etcd/clientv3"
)func main() {config := clientv3.Config{Endpoints:   []string{"192.168.0.113:2379"}, // 集群列表DialTimeout: 5 * time.Second,}// 建立一個客戶端client, err := clientv3.New(config)if err != nil {fmt.Println(err)return}// lease實現鎖自動過期:// op操作// txn事務: if else then// 1, 上鎖 (創建租約, 自動續租, 拿著租約去搶占一個key)lease := clientv3.NewLease(client)// 申請一個5秒的租約leaseGrantResp, err := lease.Grant(context.TODO(), 5)if err != nil {fmt.Println(err)return}// 拿到租約的IDleaseId := leaseGrantResp.ID// 準備一個用于取消自動續租的contextctx, cancelFunc := context.WithCancel(context.TODO())// 確保函數退出后, 自動續租會停止defer cancelFunc()defer lease.Revoke(context.TODO(), leaseId)// 5秒后會取消自動續租keepRespChan, err := lease.KeepAlive(ctx, leaseId)if err != nil {fmt.Println(err)return}// 處理續約應答的協程go func() {for {select {case keepResp := <-keepRespChan:if keepResp == nil {fmt.Println("租約已經失效了")goto END} else { // 每秒會續租一次, 所以就會受到一次應答fmt.Println("收到自動續租應答:", keepResp.ID)}}}END:}()//  if 不存在key， then 設置它, else 搶鎖失敗kv := clientv3.NewKV(client)// 創建事務txn := kv.Txn(context.TODO())// 定義事務// 如果key不存在txn.If(clientv3.Compare(clientv3.CreateRevision("/demo/A/B1"), "=", 0)).Then(clientv3.OpPut("/demo/A/B1", "xxx", clientv3.WithLease(leaseId))).Else(clientv3.OpGet("/demo/A/B1")) // 否則搶鎖失敗// 提交事務txnResp, err := txn.Commit()if err != nil {fmt.Println(err)return // 沒有問題}// 判斷是否搶到了鎖if !txnResp.Succeeded {fmt.Println("鎖被占用:", string(txnResp.Responses[0].GetResponseRange().Kvs[0].Value))return}// 2, 處理業務fmt.Println("處理任務")time.Sleep(5 * time.Second)// 3, 釋放鎖(取消自動續租, 釋放租約)// defer 會把租約釋放掉, 關聯的KV就被刪除了
}

總結

以上是生活随笔為你收集整理的etcd 笔记（08）— 基于 etcd 实现分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。