基于ZooKeeper的分布式锁
ZooKeeper 里实现分布式锁的基本逻辑：
1.zookeeper中创建一个根节点（Locks），用于后续各个客户端的锁操作。
2.想要获取锁的client都在Locks中创建一个自增序的子节点，每个client得到一个序号，如果自己的序号是最小的则获得锁。
3.如果没有得到锁，就监控排在自己前面的序号节点，并且设置默认时间，等待它的释放。
4.业务操作后释放锁,然后监控自己的节点的client就被唤醒得到锁。(例如client A需要释放锁，只需要把对应的节点1删除掉，因为client B已经关注了节点1，那么当节点1被删除后，zookeeper就会通知client B：你是序号最小的了，可以获取锁了)
释放锁的过程相对比较简单，就是删除自己创建的那个子节点即可。
解决方案目录：
Demo1 Demo2为测试场景
ZooKeepr_Lock为锁操作代码
下面贴一下代码看看
public class ZooKeeprDistributedLock : IWatcher
{
/// <summary>
/// zk链接字符串
/// </summary>
private String connectString = "127.0.0.1:2181";
private ZooKeeper zk;
private string root = "/locks"; //根
private string lockName; //竞争资源的标志
private string waitNode; //等待前一个锁
private string myZnode; //当前锁
private AutoResetEvent autoevent;
private TimeSpan sessionTimeout = TimeSpan.FromMilliseconds();
private IList<Exception> exception = new List<Exception>(); /// <summary>
/// 创建分布式锁
/// </summary>
/// <param name="lockName">竞争资源标志,lockName中不能包含单词lock</param>
public ZooKeeprDistributedLock(string lockName)
{
this.lockName = lockName;
// 创建一个与服务器的连接
try
{
zk = new ZooKeeper(connectString, sessionTimeout, this);
Stopwatch sw = new Stopwatch();
sw.Start();
while (true)
{
if (zk.State == States.CONNECTING) { break; }
if (zk.State == States.CONNECTED) { break; }
}
sw.Stop();
TimeSpan ts2 = sw.Elapsed;
Console.WriteLine("zoo连接总共花费{0}ms.", ts2.TotalMilliseconds); var stat = zk.Exists(root, false);
if (stat == null)
{
// 创建根节点
zk.Create(root, new byte[], Ids.OPEN_ACL_UNSAFE, CreateMode.Persistent);
}
}
catch (KeeperException e)
{
throw e;
}
} /// <summary>
/// zookeeper节点的监视器
/// </summary>
public virtual void Process(WatchedEvent @event) {
if (this.autoevent != null)
{
//将事件状态设置为终止状态，允许一个或多个等待线程继续；如果该操作成功，则返回true；否则，返回false
this.autoevent.Set();
}
} public virtual bool tryLock()
{
try
{
string splitStr = "_lock_";
if (lockName.Contains(splitStr))
{
//throw new LockException("lockName can not contains \\u000B");
}
//创建临时子节点
myZnode = zk.Create(root + "/" + lockName + splitStr, new byte[], Ids.OPEN_ACL_UNSAFE, CreateMode.EphemeralSequential);
Console.WriteLine(myZnode + " 创建完成！ ");
//取出所有子节点
IList<string> subNodes = zk.GetChildren(root, false).ToList<string>();
//取出所有lockName的锁
IList<string> lockObjNodes = new List<string>();
foreach (string node in subNodes)
{
if (node.StartsWith(lockName))
{
lockObjNodes.Add(node);
}
}
Array alockObjNodes = lockObjNodes.ToArray();
Array.Sort(alockObjNodes);
Console.WriteLine(myZnode + "==" + lockObjNodes[]);
if (myZnode.Equals(root + "/" + lockObjNodes[]))
{
//如果是最小的节点,则表示取得锁
Console.WriteLine(myZnode + " 获取锁成功！ ");
return true;
}
//如果不是最小的节点，找到比自己小1的节点
string subMyZnode = myZnode.Substring(myZnode.LastIndexOf("/", StringComparison.Ordinal) + );
waitNode = lockObjNodes[Array.BinarySearch(alockObjNodes, subMyZnode) - ];
}
catch (KeeperException e)
{
throw e;
}
return false;
} public virtual bool tryLock(TimeSpan time)
{
try
{
if (this.tryLock())
{
return true;
}
return waitForLock(waitNode, time);
}
catch (KeeperException e)
{
throw e;
}
} /// <summary>
/// 等待锁
/// </summary>
/// <param name="lower">需等待的锁节点</param>
/// <param name="waitTime">等待时间</param>
/// <returns></returns>
private bool waitForLock(string lower, TimeSpan waitTime)
{
var stat = zk.Exists(root + "/" + lower, true);
//判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
if (stat != null)
{
Console.WriteLine("Thread " + System.Threading.Thread.CurrentThread.Name + " waiting for " + root + "/" + lower);
autoevent = new AutoResetEvent(false);
//阻止当前线程，直到当前实例收到信号，使用 TimeSpan 度量时间间隔并指定是否在等待之前退出同步域
bool r = autoevent.WaitOne(waitTime);
autoevent.Dispose();
autoevent = null;
return r;
}
else return true;
} /// <summary>
/// 解除锁
/// </summary>
public virtual void unlock()
{
try
{
Console.WriteLine("unlock " + myZnode);
zk.Delete(myZnode, -);
myZnode = null;
zk.Dispose();
}
catch (KeeperException e)
{
throw e;
}
}
}
ZooKeeprDistributedLock
然后先看demo2 : 当前获取到锁以后 释放锁的操作被阻塞 然后运行demo1 进行测试
int count = ;//库存 商品编号1079233
if (count == )
{
ZooKeeprDistributedLock zklock = new ZooKeeprDistributedLock("Getorder_Pid1079233");
//创建锁
if (zklock.tryLock(TimeSpan.FromMilliseconds()))
{
Console.WriteLine("Demo2创建订单成功！");
}
else { Console.WriteLine("Demo2创建订单失败了！"); }
Thread.Sleep();//对操作释放锁进行阻塞
Console.WriteLine(DateTime.Now.ToString("yyyyMMdd HH:mm:ss")); //要进行释放锁的操作时间 主要测试当前锁释放后 Demo1的节点监控是否唤起
zklock.unlock();//释放锁
Console.ReadKey();
}
demo1：demo1会对排在前面的节点进行监控 当demo2释放锁后 demo1获取锁 demo1创建订单与释放锁之间打印了操作时间
可以跟demo2进行释放锁的时间进行对比下
int count = ;//库存 商品编号1079233
if (count == )
{
ZooKeeprDistributedLock zklock = new ZooKeeprDistributedLock("Getorder_Pid1079233");
if (zklock.tryLock(TimeSpan.FromMilliseconds()))
{
Console.WriteLine("Demo1创建订单成功！");
}
else
{
Console.WriteLine("Demo1创建订单失败了！");
}
Console.WriteLine(DateTime.Now.ToString("yyyyMMdd HH:mm:ss"));
zklock.unlock();
Console.ReadKey();
}
这里是我运行后的结果,只精确到秒，可以看到demo2释放锁后，demo1的AutoResetEvent立即被阻断了然后demo1也就获得了锁！
场景二：将demo1的监听注释后，demo2未释放锁，demo1创建订单失败
//if (zklock.tryLock(TimeSpan.FromMilliseconds(50000)))
if (zklock.tryLock())
有关此篇一些图片及内容借鉴了几位园友的博文，在此感谢！