今天就跟大家聊聊有關(guān)使用ZooKeeper怎么實(shí)現(xiàn)一個(gè)分布式鎖，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

創(chuàng)新互聯(lián)專注于紅橋網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗(yàn)。 熱誠為您提供紅橋營銷型網(wǎng)站建設(shè)，紅橋網(wǎng)站制作、紅橋網(wǎng)頁設(shè)計(jì)、紅橋網(wǎng)站官網(wǎng)定制、成都微信小程序服務(wù)，打造紅橋網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供紅橋網(wǎng)站排名全網(wǎng)營銷落地服務(wù)。

ZooKeeper 是一個(gè)典型的分布式數(shù)據(jù)一致性解決方案，分布式應(yīng)用程序可以基于 ZooKeeper 實(shí)現(xiàn)諸如數(shù)據(jù)發(fā)布/訂閱、負(fù)載均衡、分布式協(xié)調(diào)/通知、集群管理、Master 選舉、分布式鎖等功能。

節(jié)點(diǎn)

在介紹 ZooKeeper 分布式鎖前需要先了解一下 ZooKeeper 中節(jié)點(diǎn)（Znode），ZooKeeper 的數(shù)據(jù)存儲數(shù)據(jù)模型是一棵樹（Znode Tree)，由斜杠（/）的進(jìn)行分割的路徑，就是一個(gè) Znode（如 /locks/my_lock）。每個(gè) Znode 上都會保存自己的數(shù)據(jù)內(nèi)容，同時(shí)還會保存一系列屬性信息。

Znode 又分為以下四種類型：

鎖原理

ZooKeeper 分布式鎖是基于 臨時(shí)順序節(jié)點(diǎn)來實(shí)現(xiàn)的，鎖可理解為 ZooKeeper 上的一個(gè)節(jié)點(diǎn)，當(dāng)需要獲取鎖時(shí)，就在這個(gè)鎖節(jié)點(diǎn)下創(chuàng)建一個(gè)臨時(shí)順序節(jié)點(diǎn)。當(dāng)存在多個(gè)客戶端同時(shí)來獲取鎖，就按順序依次創(chuàng)建多個(gè)臨時(shí)順序節(jié)點(diǎn)，但只有排列序號是第一的那個(gè)節(jié)點(diǎn)能獲取鎖成功，其他節(jié)點(diǎn)則按順序分別監(jiān)聽前一個(gè)節(jié)點(diǎn)的變化，當(dāng)被監(jiān)聽者釋放鎖時(shí)，監(jiān)聽者就可以馬上獲得鎖。

而且用臨時(shí)順序節(jié)點(diǎn)的另外一個(gè)用意是如果某個(gè)客戶端創(chuàng)建臨時(shí)順序節(jié)點(diǎn)后，自己意外宕機(jī)了也沒關(guān)系，ZooKeeper 感知到某個(gè)客戶端宕機(jī)后會自動刪除對應(yīng)的臨時(shí)順序節(jié)點(diǎn)，相當(dāng)于自動釋放鎖。

如上圖：ClientA 和 ClientB 同時(shí)想獲取鎖，所以都在 locks 節(jié)點(diǎn)下創(chuàng)建了一個(gè)臨時(shí)節(jié)點(diǎn) 1 和 2，而 1 是當(dāng)前 locks 節(jié)點(diǎn)下排列序號第一的節(jié)點(diǎn)，所以 ClientA 獲取鎖成功，而 ClientB 處于等待狀態(tài)，這時(shí) ZooKeeper 中的 2 節(jié)點(diǎn)會監(jiān)聽 1 節(jié)點(diǎn)，當(dāng) 1節(jié)點(diǎn)鎖釋放（節(jié)點(diǎn)被刪除）時(shí)，2 就變成了 locks 節(jié)點(diǎn)下排列序號第一的節(jié)點(diǎn)，這樣 ClientB 就獲取鎖成功了。

代碼測試

請確保 ZooKeeper 服務(wù)已啟動，ZooKeeper 的搭建可參考Kafka 集群 中的 ZooKeeper 集群部分

以下是基于 C# 的測試，Java 可使用 Curator 框架，實(shí)現(xiàn)原理和上面描述是一致的，有興趣可以看看源碼，應(yīng)該也不難理解。

創(chuàng)建 .NET Core 控制臺程序 Nuget

安裝 ZooKeeperNetEx.Recipes

創(chuàng)建 ZooKeeper Client

private const int CONNECTION_TIMEOUT = 50000;
private const string CONNECTION_STRING = "127.0.0.1:2181";
private ZooKeeper CreateClient()
{
	var zooKeeper = new ZooKeeper(CONNECTION_STRING, CONNECTION_TIMEOUT, NullWatcher.Instance);
	Stopwatch sw = new Stopwatch();
	sw.Start();
	while (sw.ElapsedMilliseconds < CONNECTION_TIMEOUT)
	{
		var state = zooKeeper.getState();
		if (state == ZooKeeper.States.CONNECTED || state == ZooKeeper.States.CONNECTING)
		{
			break;
		}
	}
	sw.Stop();
	return zooKeeper;
}

class NullWatcher : Watcher
  {
    public static readonly NullWatcher Instance = new NullWatcher();
    private NullWatcher() { }
    public override Task process(WatchedEvent @event)
    {
      return Task.CompletedTask;
    }
  }

添加 Lock 方法

/// 

/// 加鎖
/// 
/// 加鎖的節(jié)點(diǎn)名
/// 加鎖成功后需要執(zhí)行的邏輯
/// 鎖釋放后需要執(zhí)行的邏輯，可為空
/// 
public async Task Lock(string key, Action lockAcquiredAction, Action lockReleasedAction = null)
{
	// 獲取 ZooKeeper Client
	ZooKeeper keeper = CreateClient();
	// 指定鎖節(jié)點(diǎn)
	WriteLock writeLock = new WriteLock(keeper, $"/{key}", null);

	var lockCallback = new LockCallback(() =>
	{
		lockAcquiredAction.Invoke();
		writeLock.unlock();
	}, lockReleasedAction);
	// 綁定鎖獲取和釋放的監(jiān)聽對象
	writeLock.setLockListener(lockCallback);
	// 獲取鎖（獲取失敗時(shí)會監(jiān)聽上一個(gè)臨時(shí)節(jié)點(diǎn)）
	await writeLock.Lock();
}

class LockCallback : LockListener
{
	private readonly Action _lockAcquiredAction;
	private readonly Action _lockReleasedAction;

	public LockCallback(Action lockAcquiredAction, Action lockReleasedAction)
	{
		_lockAcquiredAction = lockAcquiredAction;
		_lockReleasedAction = lockReleasedAction;
	}

	/// 

	/// 獲取鎖成功回調(diào)
	/// 
	/// 
	public Task lockAcquired()
	{
		_lockAcquiredAction?.Invoke();
		return Task.FromResult(0);
	}

	/// 

	/// 釋放鎖成功回調(diào)
	/// 
	/// 
	public Task lockReleased()
	{
		_lockReleasedAction?.Invoke();
		return Task.FromResult(0);
	}
}

多線程模擬測試

static async Task RunAsync()
{
	Parallel.For(1, 10, async (i) =>
	{
		await new ZooKeeprDistributedLock().Lock("locks", () =>
		{
			Console.WriteLine($"第{i}個(gè)請求,獲取鎖成功:{DateTime.Now},線程Id:{Thread.CurrentThread.ManagedThreadId}");
			Thread.Sleep(1000); // 業(yè)務(wù)邏輯...
		}, () =>
		{
			Console.WriteLine($"第{i}個(gè)請求,釋放鎖成功:{DateTime.Now},線程Id:{Thread.CurrentThread.ManagedThreadId}");
			Console.WriteLine("-------------------------------");
		});
	});
	await Task.CompletedTask;
}

看完上述內(nèi)容，你們對使用ZooKeeper怎么實(shí)現(xiàn)一個(gè)分布式鎖有進(jìn)一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝大家的支持。