Memcached 和 Redis 分布式鎖方案

分布式緩存，能解決單台服務器內存不能無限擴張的瓶頸。在分布式緩存的應用中，會遇到多個客戶端同時爭用的問題。這個時候，需要用到分布式鎖，得到鎖的客戶端才有操作權限。

Memcached 和 Redis 是常用的分布式緩存構建方案，下面列舉下基於Memcached 和 Redis 分布式鎖的實現方法。

Memcached 分布式鎖

Memcached 可以使用 add 命令，該命令只有KEY不存在時，才進行添加，或者不會處理。Memcached 所有命令都是原子性的，並發下add 同一個KEY ，只會一個會成功。

利用這個原理，可以先定義一個 鎖 LockKEY ，add 成功的認為是得到鎖。並且設置[過期超時] 時間，保證宕機后，也不會死鎖。

在具體操作完后，判斷是否此次操作已超時。如果超時則不刪除鎖，如果不超時則刪除鎖。

偽代碼：

         if (mc.Add("LockKey", "Value", expiredtime))
            {
                //得到鎖
                try
                {
                    //do business  function

                    //檢查超時
                    if (!CheckedTimeOut())
                    {
                        mc.Delete("LockKey");
                    }
                }
                catch (Exception e)
                {
                    mc.Delete("LockKey");
                }
               
            }

 

Redis 分布式鎖

Redis  沒有add 命令，但有SETNX（SET if Not eXists）若給定的 key 已經存在，則 SETNX不做任何動作。設置成功，返回 1 。設置失敗，返回 0 。

SETNX 命令不能設置過期時間，需要再使用 EXPIRE 命令設置過期時間。

偽代碼：

            int lockResult = rd.SETNX("LockKey", "Value");
            if (lockResult == 1)
            {
                //[1]得到鎖

                //[2]設置超時過期時間
                rd.EXPIRE("LockKey", expiredtime);

                try
                {
                    //do business  function

                    //檢查超時
                    if (!CheckedTimeOut())
                    {
                        rd.DEL("LockKey");
                    }
                }
                catch (Exception e)
                {
                    rd.DEL("LockKey");
                }

            }

　  這種做法，有一個很大的潛在風險。[1]得到鎖后，再執行[2] 設置過期時間。如果在這期間出現宕機，則會導致沒有設置過期時間。按Redis 的默認緩存過期策略，這個鎖將不會釋放，產生死鎖。

所以不推薦用這種做法，應該用其它方式來實現鎖的超時過期策略：

     1：SETNX  value 值=當前時間+過期超時時間，返回1 則獲得鎖，返回0則沒有獲得鎖。轉2。

     2：GET 獲取 value 的值 。判斷鎖是否過期超時。如果超時，轉3。

     3：GETSET（將給定 key 的值設為 value ，並返回 key 的舊值），GETSET  value 值=當前時間+過期超時時間， 判斷得到的value 如果仍然是超時的，那就說明得到鎖，否則沒有得到鎖。

從2並發進到3 的操作，會多次改寫超時時間，但這個不會有什么影響。

偽代碼：

 

            string expiredtime = DateTime.Now.AddMinutes(LockTimeoutMinutes).ToString();
            int lockResult = rd.SETNX("LockKey", expiredtime);
            bool getLock = false;
            if (lockResult == 1)
            {
                //得到鎖
                getLock = true;
            }
            else
            {
                string curExpiredtime = rd.GET("LockKey");

                //檢查鎖超時
                if (CheckedLockTimeOut(expiredtime))
                {
                    expiredtime = DateTime.Now.AddMinutes(LockTimeoutMinutes).ToString();
                    string newExpiredTime = GETSET(expiredtime);
                    if (CheckedLockTimeOut(newExpiredTime))
                    {
                        //得到鎖
                        getLock = true;
                    }
                }
            }
            if (getLock)
            {
                try
                {
                    //do business  function

                    //檢查超時
                    if (!CheckedTimeOut())
                    {
                        rd.DEL("LockKey");
                    }
                }
                catch (Exception e)
                {
                    rd.DEL("LockKey");
                }
            }

個人覺得這種做法，還是不完美。　

ZooKeeper的分布式鎖，下篇再學習探討。

 

參考：

鄭昀，  電商課題V：分布式鎖

jeffkit，  用Redis實現分布式鎖