Csharp/C#教程：ZooKeeper 实现分布式锁的方法示例分享

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

节点

在介绍ZooKeeper分布式锁前需要先了解一下ZooKeeper中节点（Znode），ZooKeeper的数据存储数据模型是一棵树（ZnodeTree)，由斜杠（/）的进行分割的路径，就是一个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框架，实现原理和上面描述是一致的，有兴趣可以看看源码，应该也不难理解。

创建.NETCore控制台程序Nuget

安装ZooKeeperNetEx.Recipes

创建ZooKeeperClient

privateconstintCONNECTION_TIMEOUT=50000; privateconststringCONNECTION_STRING="127.0.0.1:2181"; privateZooKeeperCreateClient() { varzooKeeper=newZooKeeper(CONNECTION_STRING,CONNECTION_TIMEOUT,NullWatcher.Instance); Stopwatchsw=newStopwatch(); sw.Start(); while(sw.ElapsedMilliseconds<CONNECTION_TIMEOUT) { varstate=zooKeeper.getState(); if(state==ZooKeeper.States.CONNECTED||state==ZooKeeper.States.CONNECTING) { break; } } sw.Stop(); returnzooKeeper; } classNullWatcher:Watcher { publicstaticreadonlyNullWatcherInstance=newNullWatcher(); privateNullWatcher(){} publicoverrideTaskprocess(WatchedEvent@event) { returnTask.CompletedTask; } }

添加Lock方法

///<summary> ///加锁 ///</summary> ///<paramname="key">加锁的节点名</param> ///<paramname="lockAcquiredAction">加锁成功后需要执行的逻辑</param> ///<paramname="lockReleasedAction">锁释放后需要执行的逻辑，可为空</param> ///<returns></returns> publicasyncTaskLock(stringkey,ActionlockAcquiredAction,ActionlockReleasedAction=null) { //获取ZooKeeperClient ZooKeeperkeeper=CreateClient(); //指定锁节点 WriteLockwriteLock=newWriteLock(keeper,$"/{key}",null); varlockCallback=newLockCallback(()=> { lockAcquiredAction.Invoke(); writeLock.unlock(); },lockReleasedAction); //绑定锁获取和释放的监听对象 writeLock.setLockListener(lockCallback); //获取锁（获取失败时会监听上一个临时节点） awaitwriteLock.Lock(); } classLockCallback:LockListener { privatereadonlyAction_lockAcquiredAction; privatereadonlyAction_lockReleasedAction; publicLockCallback(ActionlockAcquiredAction,ActionlockReleasedAction) { _lockAcquiredAction=lockAcquiredAction; _lockReleasedAction=lockReleasedAction; } ///<summary> ///获取锁成功回调 ///</summary> ///<returns></returns> publicTasklockAcquired() { _lockAcquiredAction?.Invoke(); returnTask.FromResult(0); } ///<summary> ///释放锁成功回调 ///</summary> ///<returns></returns> publicTasklockReleased() { _lockReleasedAction?.Invoke(); returnTask.FromResult(0); } }

多线程模拟测试

staticasyncTaskRunAsync() { Parallel.For(1,10,async(i)=> { awaitnewZooKeeprDistributedLock().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("-------------------------------"); }); }); awaitTask.CompletedTask; }

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

您可能感兴趣的文章:浅谈Java(SpringBoot)基于zookeeper的分布式锁实现浅谈分布式锁的几种使用方式（redis、zookeeper、数据库）zookeeper实现分布式锁如何操作Redis和zookeeper实现分布式锁java使用zookeeper实现的分布式锁示例

ZooKeeper的安装及部署教程

C#窗体间常用的几种传值方式及委托与事件详解

上述就是C#学习教程：ZooKeeper 实现分布式锁的方法示例分享的全部内容，如果对大家有所用处且需要了解更多关于C#学习教程，希望大家多多关注—计算机技术网(www.ctvol.com)!