分布式锁
经常用于在解决分布式环境下的业务一致性和协调分布式环境。
实际业务场景中，比如说解决并发一瞬间的重复下单，重复确认收货，重复发现金券等。
使用分布式锁的场景一般不能太多。
开源地址：http://git.oschina.net/chejiangyi/XXF.BaseService.DistributedLock
开源相关群: .net 开源基础服务 238543768
这里整理了C#.net关于redis分布式锁和zookeeper分布式锁的实现，仅用于研究。（可能有bug）
采用ServiceStack.Redis实现Redis分布式锁
/** Redis分布式锁
* 采用ServiceStack.Redis实现的Redis分布式锁
* 详情可阅读其开源代码
* 备注：不同版本的 ServiceStack.Redis 实现reidslock机制不同 xxf里面默认使用2.2版本*/publicclassRedisDistributedLock : BaseRedisDistributedLock
{PRivateServiceStack.Redis.RedisLock _lock;privateRedisClient _client;publicRedisDistributedLock(stringredisserver,stringkey)
:base(redisserver, key)
{
}publicoverrideLockResult TryGetDistributedLock(TimeSpan? getlockTimeOut, TimeSpan?taskrunTimeOut)
{if(lockresult ==LockResult.Success)thrownewDistributedLockException("检测到当前锁已获取");
_client=DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient();/** 阅读源码发现当其获取锁后,redis连接资源会一直占用,知道获取锁的资源释放后,连接才会跳出，可能会导致连接池资源的浪费。*/
try{this._lock =newServiceStack.Redis.RedisLock(_client, key, getlockTimeOut);
lockresult=LockResult.Success;
}catch(Exception exp)
{
XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁系统级别严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
lockresult=LockResult.LockSystemExceptionFailure;
}returnlockresult;
}publicoverridevoidDispose()
{try{if(this._lock !=null)this._lock.Dispose();if(_client !=null)this._client.Dispose();
}catch(Exception exp)
{
XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁释放严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
}
}
}
来自网络的java实现Redis分布式锁（C#版）
/** Redis分布式锁
* 采用网络上java实现的Redis分布式锁
* 参考http://www.blogjava.net/hello-yun/archive/2014/01/15/408988.html* 详情可阅读其开源代码*/publicclassRedisDistributedLockFromJava : BaseRedisDistributedLock
{publicRedisDistributedLockFromJava(stringredisserver,stringkey)
:base(redisserver, key)
{
}publicoverrideLockResult TryGetDistributedLock(TimeSpan? getlockTimeOut, TimeSpan?taskrunTimeOut)
{if(lockresult ==LockResult.Success)thrownewDistributedLockException("检测到当前锁已获取");try{//1. 通过SETNX试图获取一个lock
string @lock = key;longtaskexpiredMilliseconds = (taskrunTimeOut !=null? (long)taskrunTimeOut.Value.TotalMilliseconds : (long)DistributedLockConfig.MaxLockTaskRunTime);longgetlockexpiredMilliseconds = (getlockTimeOut !=null? (long)getlockTimeOut.Value.TotalMilliseconds :0);longhassleepMilliseconds =0;while(true)
{using(varredisclient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
{longvalue = CurrentUnixTimeMillis() + taskexpiredMilliseconds +1;/*Java以前版本都是用SetNX,但是这种是无法设置超时时间的,不是很理解为什么,
* 可能是因为原来的redis命令比较少导致的？现在用Add不知道效果如何.
因对redis细节不了解，但个人怀疑若异常未释放锁经常发生，可能会导致内存逐步溢出*/
boolacquired = redisclient.Add<long>(@lock, value, TimeSpan.FromMilliseconds(taskexpiredMilliseconds +DistributedLockConfig.TaskLockDelayCleepUpTime));//SETNX成功，则成功获取一个锁if(acquired ==true)
{
lockresult=LockResult.Success;
}//SETNX失败，说明锁仍然被其他对象保持，检查其是否已经超时
else
{varoldValueBytes =redisclient.Get(@lock);//超时if(oldValueBytes !=null&& BitConverter.ToInt64(oldValueBytes,0) <CurrentUnixTimeMillis())
{/*此处虽然重设并获取锁,但是超时时间可能被覆盖,故重设超时时间;若有进程一直在尝试获取锁，那么锁存活时间应该被延迟*/
vargetValueBytes =redisclient.GetSet(@lock, BitConverter.GetBytes(value));varo1 = redisclient.ExpireEntryIn(@lock, TimeSpan.FromMilliseconds(taskexpiredMilliseconds + DistributedLockConfig.TaskLockDelayCleepUpTime));//这里如果程序异常终止，依然会有部分锁未释放的情况。//获取锁成功 if (getValueBytes == oldValueBytes){
lockresult=LockResult.Success;
}//已被其他进程捷足先登了else{
lockresult=LockResult.GetLockTimeOutFailure;
}
}//未超时，则直接返回失败else{
lockresult=LockResult.GetLockTimeOutFailure;
}
}
}//成功拿到锁if(lockresult ==LockResult.Success)
break;//获取锁超时if(hassleepMilliseconds >=getlockexpiredMilliseconds)
{
lockresult=LockResult.GetLockTimeOutFailure;
break;
}//继续等待System.Threading.Thread.Sleep(DistributedLockConfig.GetLockFailSleepTime);
hassleepMilliseconds+=DistributedLockConfig.GetLockFailSleepTime;
}
}catch(Exception exp)
{
XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁系统级别严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
lockresult=LockResult.LockSystemExceptionFailure;
}returnlockresult;
}privatelongCurrentUnixTimeMillis()
{return(long)(System.DateTime.UtcNow -newSystem.DateTime(1970,1,1,0,0,0, System.DateTimeKind.Utc)).TotalMilliseconds;
}publicoverridevoidDispose()
{if(lockresult == LockResult.Success || lockresult ==LockResult.LockSystemExceptionFailure)
{try{longcurrent =CurrentUnixTimeMillis();using(varredisclient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
{varv =redisclient.Get(key);if(v !=null)
{//避免删除非自己获取得到的锁if(current < BitConverter.ToInt64(v,0))
{
redisclient.Del(key);
}
}
}
}catch(Exception exp)
{
XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁释放严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
}
}
}
}
ServiceStack.Redis内部实现版本（较旧）
/** Redis分布式锁
* 采用ServiceStack.Redis实现的Redis分布式锁
* 详情可阅读其开源代码
* 备注：不同版本的 ServiceStack.Redis 实现reidslock机制不同
* 拷贝自网络开源代码 较旧的实现版本*/publicclassRedisDistributedLockFromServiceStack : BaseRedisDistributedLock
{publicRedisDistributedLockFromServiceStack(stringredisserver,stringkey)
:base(redisserver, key)
{
}publicoverrideLockResult TryGetDistributedLock(TimeSpan? getlockTimeOut, TimeSpan?taskrunTimeOut)
{if(lockresult ==LockResult.Success)thrownewDistributedLockException("检测到当前锁已获取");try{using(varredisClient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
{
ExecExtensions.RetryUntilTrue(
()=>{//This pattern is taken from the redis command for SETNXhttp://redis.io/commands/setnx//Calculate a unix time for when the lock should expire
TimeSpan realSpan = taskrunTimeOut ?? TimeSpan.FromMilliseconds(DistributedLockConfig.MaxLockTaskRunTime);//new TimeSpan(365, 0, 0, 0);//if nothing is passed in the timeout hold for a year DateTime expireTime = DateTime.UtcNow.Add(realSpan);stringlockString = (expireTime.ToUnixTimeMs() +1).ToString();//Try to set the lock, if it does not exist this will succeed and the lock is obtained
var nx = redisClient.SetEntryIfNotExists(key, lockString);if(nx)
{
lockresult=LockResult.Success;returntrue;
}//If we've gotten here then a key for the lock is present. This could be because the lock is
//correctly acquired or it could be because a client that had acquired the lock crashed (or didn't release it properly).
//Therefore we need to get the value of the lock to see when it should expireredisClient.Watch(key);stringlockExpireString = redisClient.Get<string>(key);longlockExpireTime;if(!long.TryParse(lockExpireString,outlockExpireTime))
{
redisClient.UnWatch();//since the client is scoped externally
lockresult = LockResult.GetLockTimeOutFailure;returnfalse;
}//If the expire time is greater than the current time then we can't let the lock go yet
if (lockExpireTime > DateTime.UtcNow.ToUnixTimeMs()){
redisClient.UnWatch();//since the client is scoped externally
lockresult = LockResult.GetLockTimeOutFailure;returnfalse;
}//If the expire time is less than the current time then it wasn't released properly and we can attempt to//acquire the lock. The above call to Watch(_lockKey) enrolled the key in monitoring, so if it changes//before we call Commit() below, the Commit will fail and return false, which means that another thread//was able to acquire the lock before we finished processing. using (var trans = redisClient.CreateTransaction())//we started the "Watch" above; this tx will succeed if the value has not moved {trans.QueueCommand(r =>r.Set(key, lockString));//return trans.Commit();//returns false if Transaction failed var t = trans.Commit();if(t ==false)
lockresult=LockResult.GetLockTimeOutFailure;elselockresult=LockResult.Success;returnt;
}
},
getlockTimeOut
);
}
}catch(Exception exp)
{
XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁系统级别严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
lockresult=LockResult.LockSystemExceptionFailure;
}returnlockresult;
}publicoverridevoidDispose()
{try{using(varredisClient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
{
redisClient.Remove(key);
}
}catch(Exception exp)
{
XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁释放严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
}
}
}
Zookeeper 版本实现分布式锁
/** 来源java网络源码的zookeeper分布式锁实现（目前仅翻译并简单测试ok，未来集成入sdk）
* 备注: 共享锁在同一个进程中很容易实现，但是在跨进程或者在不同 Server 之间就不好实现了。Zookeeper 却很容易实现这个功能，实现方式也是需要获得锁的 Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点，
然后调用 getChildren方法获取当前的目录节点列表中最小的目录节点是不是就是自己创建的目录节点，如果正是自己创建的，那么它就获得了这个锁，
如果不是那么它就调用 exists(String path, boolean watch) 方法并监控 Zookeeper 上目录节点列表的变化，一直到自己创建的节点是列表中最小编号的目录节点，
从而获得锁，释放锁很简单，只要删除前面它自己所创建的目录节点就行了。*/publicclassZooKeeprDistributedLockFromJava : IWatcher
{privateZooKeeper zk;privatestringroot ="/locks";//根privatestringlockName;//竞争资源的标志privatestringwaitNode;//等待前一个锁privatestringmyZnode;//当前锁//private CountDownLatch latch;//计数器privateAutoResetEvent autoevent;privateTimeSpansessionTimeout = TimeSpan.FromMilliseconds(30000);privateIList<Exception> exception =newList<Exception>();///<summary>///创建分布式锁,使用前请确认config配置的zookeeper服务可用</summary>///<param name="config">127.0.0.1:2181</param>///<param name="lockName">竞争资源标志,lockName中不能包含单词lock</param>publicZooKeeprDistributedLockFromJava(stringconfig,stringlockName)
{this.lockName =lockName;//创建一个与服务器的连接try
{
zk=newZooKeeper(config, sessionTimeout,this);varstat = zk.Exists(root,false);if(stat ==null)
{//创建根节点zk.Create(root,newbyte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.Persistent);
}
}catch(KeeperException e)
{throwe;
}
}///<summary>///zookeeper节点的监视器///</summary>publicvirtualvoidProcess(WatchedEvent @event)
{if(this.autoevent !=null)
{this.autoevent.Set();
}
}publicvirtualbooltryLock()
{try{stringsplitStr ="_lock_";if(lockName.Contains(splitStr))
{//throw new LockException("lockName can not contains \\u000B");
}//创建临时子节点myZnode = zk.Create(root +"/"+ lockName + splitStr,newbyte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EphemeralSequential);
Console.WriteLine(myZnode+"is created");//取出所有子节点IList<string> subNodes = zk.GetChildren(root,false);//取出所有lockName的锁IList<string> lockObjNodes =newList<string>();foreach(stringnodeinsubNodes)
{if(node.StartsWith(lockName))
{
lockObjNodes.Add(node);
}
}
Array alockObjNodes=lockObjNodes.ToArray();
Array.Sort(alockObjNodes);
Console.WriteLine(myZnode+"=="+ lockObjNodes[0]);if(myZnode.Equals(root +"/"+ lockObjNodes[0]))
{//如果是最小的节点,则表示取得锁returntrue;
}//如果不是最小的节点，找到比自己小1的节点stringsubMyZnode = myZnode.Substring(myZnode.LastIndexOf("/", StringComparison.Ordinal) +1);
waitNode= lockObjNodes[Array.BinarySearch(alockObjNodes, subMyZnode) -1];
}catch(KeeperException e)
{throwe;
}returnfalse;
}publicvirtualbooltryLock(TimeSpan time)
{try{if(this.tryLock())
{returntrue;
}returnwaitForLock(waitNode, time);
}catch(KeeperException e)
{throwe;
}returnfalse;
}privateboolwaitForLock(stringlower, TimeSpan waitTime)
{varstat = zk.Exists(root +"/"+ lower,true);//判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听if(stat !=null)
{
Console.WriteLine("Thread"+ System.Threading.Thread.CurrentThread.Name +"waiting for"+ root +"/"+lower);
autoevent=newAutoResetEvent(false);boolr =autoevent.WaitOne(waitTime);
autoevent.Dispose();
autoevent=null;returnr;
}elsereturntrue;
}publicvirtualvoidunlock()
{try{
Console.WriteLine("unlock"+myZnode);
zk.Delete(myZnode,-1);
myZnode=null;
zk.Dispose();
}catch(KeeperException e)
{throwe;
}
}
}
以上代码仅做参考，未压测。
代码粘贴有些问题，详细请下载开源包运行研究。