.net 分布式架构之分布式锁实现

/** Redis分布式锁     
 * 采用ServiceStack.Redis实现的Redis分布式锁     
 * 详情可阅读其开源代码     
 * 备注：不同版本的 ServiceStack.Redis 实现reidslock机制不同 xxf里面默认使用2.2版本     
 */    public classRedisDistributedLock : BaseRedisDistributedLock
    {
        privateServiceStack.Redis.RedisLock _lock;
        privateRedisClient _client;
        public RedisDistributedLock(string redisserver, stringkey)
            : base(redisserver, key)
        {
 
        }
 
        public override LockResult TryGetDistributedLock(TimeSpan? getlockTimeOut, TimeSpan?taskrunTimeOut)
        {
            if (lockresult ==LockResult.Success)
                throw new DistributedLockException("检测到当前锁已获取");
            _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;
        }
 
        public override voidDispose()
        {
            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);
            }
        }
    }

/** Redis分布式锁     
* 采用网络上java实现的Redis分布式锁     
* 参考 http://www.blogjava.net/hello-yun/archive/2014/01/15/408988.html* 详情可阅读其开源代码     
*/    public classRedisDistributedLockFromJava : BaseRedisDistributedLock
    {
 
 
        public RedisDistributedLockFromJava(string redisserver, stringkey)
            : base(redisserver, key)
        {
 
 
        }
 
        public override LockResult TryGetDistributedLock(TimeSpan? getlockTimeOut, TimeSpan?taskrunTimeOut)
        {
            if (lockresult ==LockResult.Success)
                throw new DistributedLockException("检测到当前锁已获取");
            try{
                //1. 通过SETNX试图获取一个lock               
string @lock = key;
                long taskexpiredMilliseconds = (taskrunTimeOut != null ? (long)taskrunTimeOut.Value.TotalMilliseconds : (long)DistributedLockConfig.MaxLockTaskRunTime);
                long getlockexpiredMilliseconds = (getlockTimeOut != null ? (long)getlockTimeOut.Value.TotalMilliseconds : 0);
                long hassleepMilliseconds = 0;
                while (true)
                {
                    using (var redisclient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
                    {
                        long value = CurrentUnixTimeMillis() + taskexpiredMilliseconds + 1;
                        /*Java以前版本都是用SetNX,但是这种是无法设置超时时间的,不是很理解为什么,                         
* 可能是因为原来的redis命令比较少导致的？现在用Add不知道效果如何.                         
因对redis细节不了解，但个人怀疑若异常未释放锁经常发生，可能会导致内存逐步溢出*/                        
bool acquired = redisclient.Add<long>(@lock, value, TimeSpan.FromMilliseconds(taskexpiredMilliseconds +DistributedLockConfig.TaskLockDelayCleepUpTime));
                        //SETNX成功，则成功获取一个锁                        
                        if (acquired == true)
                        {
                            lockresult =LockResult.Success;
                        }
                        //SETNX失败，说明锁仍然被其他对象保持，检查其是否已经超时                       
else                       
{
                            var oldValueBytes =redisclient.Get(@lock);
                            //超时                            
                            if (oldValueBytes != null && BitConverter.ToInt64(oldValueBytes, 0) <CurrentUnixTimeMillis())
                            {
                                /*此处虽然重设并获取锁,但是超时时间可能被覆盖,故重设超时时间;若有进程一直在尝试获取锁，那么锁存活时间应该被延迟*/                                
var getValueBytes =redisclient.GetSet(@lock, BitConverter.GetBytes(value));
                                var o1 = 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;
        }
 
        private longCurrentUnixTimeMillis()
        {
            return (long)(System.DateTime.UtcNow - new System.DateTime(1970, 1, 1, 0, 0, 0, System.DateTimeKind.Utc)).TotalMilliseconds;
        }
 
        public override voidDispose()
        {
            if (lockresult == LockResult.Success || lockresult ==LockResult.LockSystemExceptionFailure)
            {
                try{
                    long current =CurrentUnixTimeMillis();
                    using (var redisclient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
                    {
                        var v =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);
                }
            }
        }
    }

/** Redis分布式锁    
 * 采用ServiceStack.Redis实现的Redis分布式锁    
 * 详情可阅读其开源代码    
 * 备注：不同版本的 ServiceStack.Redis 实现reidslock机制不同     
  * 拷贝自网络开源代码 较旧的实现版本    
  */    public classRedisDistributedLockFromServiceStack : BaseRedisDistributedLock
    {
        public RedisDistributedLockFromServiceStack(string redisserver, stringkey)
            : base(redisserver, key)
        {
 
 
        }
        public override LockResult TryGetDistributedLock(TimeSpan? getlockTimeOut, TimeSpan?taskrunTimeOut)
        {
            if (lockresult ==LockResult.Success)
                throw new DistributedLockException("检测到当前锁已获取");
            try{
                using (var redisClient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
                {
                    ExecExtensions.RetryUntilTrue(
                             () =>{
                                 //This pattern is taken from the redis command for SETNX http://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);
                                 string lockString = (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;
                                     return true;
                                 }
 
                                 //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 expire
redisClient.Watch(key);
                                 string lockExpireString = redisClient.Get<string>(key);
                                 longlockExpireTime;
                                 if (!long.TryParse(lockExpireString, outlockExpireTime))
                                 {
                                     redisClient.UnWatch();  //since the client is scoped externally                                     
lockresult = LockResult.GetLockTimeOutFailure;
                                     return false;
                                 }
 
                                 //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;
                                     return false;
                                 }
 
                                 //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;
        }
 
        public override voidDispose()
        {
            try{
                using (var redisClient =DistributedLockConfig.GetRedisPoolClient(redisserver).GetClient())
                {
                    redisClient.Remove(key);
                }
            }
            catch(Exception exp)
            {
                XXF.Log.ErrorLog.Write(string.Format("redis分布式尝试锁释放严重异常,redisserver:{0}", redisserver.NullToEmpty()), exp);
            }
        }
    }

/** 来源java网络源码的zookeeper分布式锁实现（目前仅翻译并简单测试ok，未来集成入sdk）  
* 备注:    共享锁在同一个进程中很容易实现，但是在跨进程或者在不同 Server 之间就不好实现了。Zookeeper 却很容易实现这个功能，实现方式也是需要获得锁的 Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点，   
然后调用 getChildren方法获取当前的目录节点列表中最小的目录节点是不是就是自己创建的目录节点，如果正是自己创建的，那么它就获得了这个锁，    
如果不是那么它就调用 exists(String path, boolean watch) 方法并监控 Zookeeper 上目录节点列表的变化，一直到自己创建的节点是列表中最小编号的目录节点，    
从而获得锁，释放锁很简单，只要删除前面它自己所创建的目录节点就行了。  
*/    public classZooKeeprDistributedLockFromJava : IWatcher
    {
        privateZooKeeper zk;
        private string root = "/locks"; //根        
        private string lockName; //竞争资源的标志        
        private string waitNode; //等待前一个锁        
        private string myZnode; //当前锁        
        //private CountDownLatch latch; //计数器        
        privateAutoResetEvent autoevent;
        private TimeSpan sessionTimeout = TimeSpan.FromMilliseconds(30000);
        private IList<Exception> exception = new List<Exception>();
 
        /// <summary>        
        ///创建分布式锁,使用前请确认config配置的zookeeper服务可用 </summary>        
        /// <param name="config">127.0.0.1:2181 </param>        
        /// <param name="lockName">竞争资源标志,lockName中不能包含单词lock </param>        
        public ZooKeeprDistributedLockFromJava(string config, stringlockName)
        {
            this.lockName =lockName;
            //创建一个与服务器的连接            
            try
{
                zk = new ZooKeeper(config, sessionTimeout, this);
                var stat = zk.Exists(root, false);
                if (stat == null)
                {
                    //创建根节点                    
                    zk.Create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.Persistent);
                }
            }
            catch(KeeperException e)
            {
                throwe;
            }
        }
 
        /// <summary>        
        ///zookeeper节点的监视器        
        /// </summary>        
        public virtual voidProcess(WatchedEvent @event)
        {
            if (this.autoevent != null)
            {
                this.autoevent.Set();
            }
        }
 
        public virtual booltryLock()
        {
            try{
                string splitStr = "_lock_";
                if(lockName.Contains(splitStr))
                {
                    //throw new LockException("lockName can not contains \u000B");                
}
                //创建临时子节点                
                myZnode = zk.Create(root + "/" + lockName + splitStr, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EphemeralSequential);
                Console.WriteLine(myZnode + "is created ");
                //取出所有子节点                
                IList<string> subNodes = zk.GetChildren(root, false);
                //取出所有lockName的锁                
                IList<string> lockObjNodes = new List<string>();
                foreach (string node insubNodes)
                {
                    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]))
                {
                    //如果是最小的节点,则表示取得锁                    
                    return true;
                }
                //如果不是最小的节点，找到比自己小1的节点               
                string subMyZnode = myZnode.Substring(myZnode.LastIndexOf("/", StringComparison.Ordinal) + 1);
                waitNode = lockObjNodes[Array.BinarySearch(alockObjNodes, subMyZnode) - 1];
            }
            catch(KeeperException e)
            {
                throwe;
            }
            return false;
        }
 
        public virtual booltryLock(TimeSpan time)
        {
            try{
                if (this.tryLock())
                {
                    return true;
                }
                returnwaitForLock(waitNode, time);
            }
            catch(KeeperException e)
            {
                throwe;
            }
            return false;
        }
 
        private bool waitForLock(stringlower, 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);
                bool r =autoevent.WaitOne(waitTime);
                autoevent.Dispose();
                autoevent = null;
                returnr;
            }
            else                return true;
        }
 
        public virtual voidunlock()
        {
            try{
                Console.WriteLine("unlock " +myZnode);
                zk.Delete(myZnode, -1);
                myZnode = null;
                zk.Dispose();
            }
            catch(KeeperException e)
            {
                throwe;
            }
        }
 
 
 
 
    }