ZooKeeper 实现分布式锁的方法示例

zookeeper 是一个典型的分布式数据一致性解决方案，分布式应用程序可以基于 zookeeper 实现诸如数据发布/订阅、负载均衡、分布式协调/通知、集群管理、master 选举、分布式锁等功能。

节点

在介绍 zookeeper 分布式锁前需要先了解一下 zookeeper 中节点（znode），zookeeper 的数据存储数据模型是一棵树（znode tree)，由斜杠（/）的进行分割的路径，就是一个 znode（如 /locks/my_lock）。每个 znode 上都会保存自己的数据内容，同时还会保存一系列属性信息。

znode 又分为以下四种类型：

类型	描述
持久节点	节点创建后，会一直存在，不会因客户端会话失效而删除
持久顺序节点	基本特性与持久节点一致，创建节点的过程中，zookeeper 会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名
临时节点	客户端会话失效或连接关闭后，该节点会被自动删除
临时顺序节点	基本特性与临时节点一致，创建节点的过程中，zookeeper 会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名

锁原理

zookeeper 分布式锁是基于 临时顺序节点 来实现的，锁可理解为 zookeeper 上的一个节点，当需要获取锁时，就在这个锁节点下创建一个临时顺序节点。当存在多个客户端同时来获取锁，就按顺序依次创建多个临时顺序节点，但只有排列序号是第一的那个节点能获取锁成功，其他节点则按顺序分别监听前一个节点的变化，当被监听者释放锁时，监听者就可以马上获得锁。

而且用临时顺序节点的另外一个用意是如果某个客户端创建临时顺序节点后，自己意外宕机了也没关系，zookeeper 感知到某个客户端宕机后会自动删除对应的临时顺序节点，相当于自动释放锁。

如上图：clienta 和 clientb 同时想获取锁，所以都在 locks 节点下创建了一个临时节点 1 和 2，而 1 是当前 locks 节点下排列序号第一的节点，所以 clienta 获取锁成功，而 clientb 处于等待状态，这时 zookeeper 中的 2 节点会监听 1 节点，当 1节点锁释放（节点被删除）时，2 就变成了 locks 节点下排列序号第一的节点，这样 clientb 就获取锁成功了。

代码测试

请确保 zookeeper 服务已启动，zookeeper 的搭建可参考kafka 集群 中的 zookeeper 集群部分

以下是基于 c# 的测试，java 可使用 curator 框架，实现原理和上面描述是一致的，有兴趣可以看看源码，应该也不难理解。

创建 .net core 控制台程序 nuget

安装 zookeepernetex.recipes

创建 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 方法

/// <summary>
/// 加锁
/// </summary>
/// <param name="key">加锁的节点名</param>
/// <param name="lockacquiredaction">加锁成功后需要执行的逻辑</param>
/// <param name="lockreleasedaction">锁释放后需要执行的逻辑，可为空</param>
/// <returns></returns>
public async task lock(string key, action lockacquiredaction, action lockreleasedaction = null)
{
	// 获取 zookeeper client
	zookeeper keeper = createclient();
	// 指定锁节点
	writelock writelock = new writelock(keeper, $"/{key}", null);

	var lockcallback = new lockcallback(() =>
	{
		lockacquiredaction.invoke();
		writelock.unlock();
	}, lockreleasedaction);
	// 绑定锁获取和释放的监听对象
	writelock.setlocklistener(lockcallback);
	// 获取锁（获取失败时会监听上一个临时节点）
	await writelock.lock();
}

class lockcallback : locklistener
{
	private readonly action _lockacquiredaction;
	private readonly action _lockreleasedaction;

	public lockcallback(action lockacquiredaction, action lockreleasedaction)
	{
		_lockacquiredaction = lockacquiredaction;
		_lockreleasedaction = lockreleasedaction;
	}

	/// <summary>
	/// 获取锁成功回调
	/// </summary>
	/// <returns></returns>
	public task lockacquired()
	{
		_lockacquiredaction?.invoke();
		return task.fromresult(0);
	}

	/// <summary>
	/// 释放锁成功回调
	/// </summary>
	/// <returns></returns>
	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}个请求,获取锁成功:{datetime.now},线程id:{thread.currentthread.managedthreadid}");
			thread.sleep(1000); // 业务逻辑...
		}, () =>
		{
			console.writeline($"第{i}个请求,释放锁成功:{datetime.now},线程id:{thread.currentthread.managedthreadid}");
			console.writeline("-------------------------------");
		});
	});
	await task.completedtask;
}

虽然模拟的是多线程并行执行，但最终都会依赖锁的获取和释放而串行执行实际业务逻辑。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。