C# 实现Zookeeper分布式锁 C# 实现Zookeeper分布式锁的参考示例
没有星星的夏季 人气:0分布式锁
互联网初期，我们系统一般都是单点部署，也就是在一台服务器完成系统的部署，后期随着用户量的增加，服务器的压力也越来越大，响应速度越来越慢，甚至出现服务器崩溃的情况。
为解决服务器压力太大，响应慢的特点，分布式系统部署出现了。
简单的说，就是我们将系统资源部署到多台服务器中，然后使用一台服务器做入口代理，根据一些决策将接收到的请求转发到资源服务器，这也就是我们常说的 反向代理（一般就是使用nginx）
虽然分布式解决了服务器压力的问题，但也带来了新的问题。
比如，我们有一个下单统计的功能，当完成下单后，需要执行统计功能，而在高访问的情况下，可能有两个下单请求（A和B）同时完成，然后一起执行了统计功能，这样可能导致的结果就是A请求未将B请求数据统计在内，而B请求可能也未将A请求数据统计在内，这样就造成了数据的统计错，这个问题的产生的根本原因就是统计功能的并发导致的，如果是单点部署的系统，我们简单的使用一个锁操作就能完成了，但是在分布式环境下，A和B请求可能同时运行在两个服务器中，普通的锁就不能起到效果了，这个时候就要使用分布式锁了。
Zookeeper分布式锁原理
分布式锁的实现发放有多种，简单的，我们可以使用数据库表去实现它，也可以使用redis去实现它，这里要使用的Zookeeper去实现分布式锁
Zookeeper分布式锁的原理是巧妙的是使用了znode临时节点的特点和监听（watcher）机制，监听机制很简单，就是我们可以给znode添加一个监听器，当znode节点状态发生改变时（如：数据内容改变，节点被删除），会通知到监听器。
前面几节介绍过znode有三种类型
PERSISTENT：持久节点，即使在创建该特定znode的客户端断开连接后，持久节点仍然存在。默认情况下，除非另有说明，否则所有znode都是持久的。
EPHEMERAL：临时节点，客户端是连接状态时，临时节点就是有效的。当客户端与ZooKeeper集合断开连接时，临时节点会自动删除。临时节点不允许有子节点。临时节点在leader选举中起着重要作用。
SEQUENTIAL：顺序节点，可以是持久的或临时的。当一个新的znode被创建为一个顺序节点时，ZooKeeper通过将10位的序列号附加到原始名称来设置znode的路径，顺序节点在锁定和同步中起重要作用。
其中，顺序节点，可以是持久的或临时的，而临时节点有个特点，就是它属于创建它的那个会话，当会话断开，临时节点就会自动删除，如果在临时节点上注册了监听器，那么监听器就会收到通知，如果临时节点有了时间顺序，那我们为实现分布式锁就又有一个想法：
假如在Zookeeper中有一个znode节点/Locker
1、当client1连接Zookeeper时，先判断/Locker节点是否存在子节点，如果没有子节点，那么会在/Locker节点下创建一个临时顺序的znode节点，假如是/client1，表示client1获取了锁状态，client1可以继续执行。
2、当client2连接Zookeeper时，先判断/Locker节点是否存在子节点，发现已经存在子节点了，然后获取/Locker下的所有子节点，同时按时间顺序排序，在最后一个节点，也就是/client1节点上注册一个监听器（watcher1），同时在/Locker节点下创建一个临时顺序的znode节点，假如是/client2。同时client2将被阻塞，而阻塞状态的释放是在监听器（watcher1）中的。
3、当client3连接Zookeeper时，先判断/Locker节点是否存在子节点，发现已经存在子节点了，然后获取/Locker下的所有子节点，同时按时间顺序排序，在最后一个节点，也就是/client2节点上注册一个监听器（watcher2），同时在/Locker节点下创建一个临时顺序的znode节点，假如是/client3。同时client2将被阻塞，而阻塞状态的释放是在监听器（watcher2）中的。
以此类推。
4、当client1执行完操作了，断开Zookeeper的连接，因为/client1是临时顺序节点，于是将会自动删除，而client2已经往/client1节点中注册了一个监听器（watcher1）,于是watcher1将会受到通知，而watcher1又会释放client2的阻塞状态。于是client2获取锁状态，继续执行。
5、当client2执行完操作了，断开Zookeeper的连接，因为/client2是临时顺序节点，于是将会自动删除，而client3已经往/client2节点中注册了一个监听器（watcher2）,于是watcher2将会受到通知，而watcher2又会释放client3的阻塞状态。于是client3获取锁状态，继续执行。
以此类推。
这样，不管分布式环境中有几台服务器，都可以保证程序的排队似的执行了。
C#实现Zookeeper分布式锁
上一节有封装过一个ZookeeperHelper的辅助类（Zookeeper基础教程（四）：C#连接使用Zookeeper），使用这个辅助类实现了一个ZookeeperLocker类：
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace AspNetCore.ZookeeperConsole
{
/// <summary>
/// 基于Zookeeper的分布式锁
/// </summary>
public class ZookeeperLocker : IDisposable
{
/// <summary>
/// 单点锁
/// </summary>
static object locker = new object();
/// <summary>
/// Zookeeper集群地址
/// </summary>
string[] address;
/// <summary>
/// Zookeeper操作辅助类
/// </summary>
ZookeeperHelper zookeeperHelper;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="lockerPath">分布式锁的根路径</param>
/// <param name="address">集群地址</param>
public ZookeeperLocker(string lockerPath, params string[] address) : this(lockerPath, 0, address)
{
}
/// <summary>
/// 构造函数
/// </summary>
/// <param name="lockerPath">分布式锁的根路径</param>
/// <param name="sessionTimeout">回话过期时间</param>
/// <param name="address">集群地址</param>
public ZookeeperLocker(string lockerPath, int sessionTimeout, params string[] address)
{
this.address = address.ToArray();
zookeeperHelper = new ZookeeperHelper(address, lockerPath);
if (sessionTimeout > 0)
{
zookeeperHelper.SessionTimeout = sessionTimeout;
}
if (!zookeeperHelper.Connect())
{
throw new Exception("connect failed:" + string.Join(",", address));
}
lock (locker)
{
if (!zookeeperHelper.Exists())//根节点不存在则创建
{
zookeeperHelper.SetData("", "", true);
}
}
}
/// <summary>
/// 生成一个锁
/// </summary>
/// <returns>返回锁名</returns>
public string CreateLock()
{
var path = Guid.NewGuid().ToString().Replace("-", "");
while (zookeeperHelper.Exists(path))
{
path = Guid.NewGuid().ToString().Replace("-", "");
}
return CreateLock(path);
}
/// <summary>
/// 使用指定的路径名称设置锁
/// </summary>
/// <param name="path">锁名，不能包含路径分隔符（/）</param>
/// <returns>返回锁名</returns>
public string CreateLock(string path)
{
if (path.Contains("/"))
{
throw new ArgumentException("invalid path");
}
return zookeeperHelper.SetData(path, "", false, true);
}
/// <summary>
/// 获取锁
/// </summary>
/// <param name="path">锁名</param>
/// <returns>如果获得锁返回true，否则一直等待</returns>
public bool Lock(string path)
{
return LockAsync(path).GetAwaiter().GetResult();
}
/// <summary>
/// 获取锁
/// </summary>
/// <param name="path">锁名</param>
/// <param name="millisecondsTimeout">超时时间，单位：毫秒</param>
/// <returns>如果获得锁返回true，否则等待指定时间后返回false</returns>
public bool Lock(string path, int millisecondsTimeout)
{
return LockAsync(path, millisecondsTimeout).GetAwaiter().GetResult();
}
/// <summary>
/// 异步获取锁等等
/// </summary>
/// <param name="path">锁名</param>
/// <returns>如果获得锁返回true，否则一直等待</returns>
public async Task<bool> LockAsync(string path)
{
return await LockAsync(path, System.Threading.Timeout.Infinite);
}
/// <summary>
/// 异步获取锁等等
/// </summary>
/// <param name="path">锁名</param>
/// <param name="millisecondsTimeout">超时时间，单位：毫秒</param>
/// <returns>如果获得锁返回true，否则等待指定时间后返回false</returns>
public async Task<bool> LockAsync(string path, int millisecondsTimeout)
{
var array = await zookeeperHelper.GetChildrenAsync("", true);
if (array != null && array.Length > 0)
{
var first = array.FirstOrDefault();
if (first == path)//正好是优先级最高的，则获得锁
{
return true;
}
var index = array.ToList().IndexOf(path);
if (index > 0)
{
//否则添加监听
var are = new AutoResetEvent(false);
var watcher = new NodeWatcher();
watcher.NodeDeleted += (ze) =>
{
are.Set();
};
if (await zookeeperHelper.WatchAsync(array[index - 1], watcher))//监听顺序节点中的前一个节点
{
if (!are.WaitOne(millisecondsTimeout))
{
return false;
}
}
are.Dispose();
}
else
{
throw new InvalidOperationException($"no locker found in path:{zookeeperHelper.CurrentPath}");
}
}
return true;
}
/// <summary>
/// 释放资源
/// </summary>
public void Dispose()
{
zookeeperHelper.Dispose();
}
}
}
现在写个程序可以模拟一下
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
namespace AspNetCore.ZookeeperConsole
{
class Program
{
static void Main(string[] args)
{
//Zookeeper连接字符串，采用host:port格式，多个地址之间使用逗号（,）隔开
string[] address = new string[] { "192.168.209.133:2181", "192.168.209.133:2181", "192.168.209.133:2181" };
//会话超时时间,单位毫秒
int sessionTimeOut = 10000;
//锁节点根路径
string lockerPath = "/Locker";
for (var i = 0; i < 10; i++)
{
string client = "client" + i;
//多线程模拟并发
new Thread(() =>
{
using (ZookeeperLocker zookeeperLocker = new ZookeeperLocker(lockerPath, sessionTimeOut, address))
{
string path = zookeeperLocker.CreateLock();
if (zookeeperLocker.Lock(path))
{
//模拟处理过程
Console.WriteLine($"【{client}】获得锁:{DateTime.Now}");
Thread.Sleep(3000);
Console.WriteLine($"【{client}】处理完成:{DateTime.Now}");
}
else
{
Console.WriteLine($"【{client}】获得锁失败:{DateTime.Now}");
}
}
}).Start();
}
Console.ReadKey();
}
}
}
运行结果如下：
可以发现，锁功能是实现了的
如果程序运行中打印日志：Client session timed out, have not heard from server in 8853ms for sessionid 0x1000000ec5500b2
或者直接抛出异常：org.apache.zookeeper.KeeperException.ConnectionLossException:“Exception_WasThrown”
只需要适当调整sessionTimeOut时间即可
加载全部内容