dubbo 經常使用的基於redis的分佈式鎖實現

 

小弟本着先會用在學習原理的原則 先用了dubbo 如今在實際業務中 由於分佈式項目作了集羣，須要用的分佈式鎖，就用到了基於redis的分佈式鎖，廢話很少說，先來代碼：

package com.tiancaibao.utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Redis distributed lock implementation.
 *	
 * @author qingzhipeng
 */
public class RedisLock {

    private static Logger logger = LoggerFactory.getLogger(RedisLock.class);
    
    private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;

    /**
     * Lock key path.
     */
    private String lockKey;

    /**
     * 鎖超時時間，防止線程在入鎖之後，無限的執行等待
     */
    private int expireMsecs = 60 * 1000;

    /**
     * 鎖等待時間，防止線程飢餓
     */
    private int timeoutMsecs = 10 * 1000;

    private volatile boolean locked = false;

    /**
     * Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
     *
     * @param lockKey lock key (ex. account:1, ...)
     */
    public RedisLock(String lockKey) {
        this.lockKey = lockKey + "_lock";
    }

    /**
     * Detailed constructor with default lock expiration of 60000 msecs.
     *
     */
    public RedisLock(String lockKey, int timeoutMsecs) {
        this(lockKey);
        this.timeoutMsecs = timeoutMsecs;
    }

    /**
     * Detailed constructor.
     *
     */
    public RedisLock(String lockKey, int timeoutMsecs, int expireMsecs) {
        this(lockKey, timeoutMsecs);
        this.expireMsecs = expireMsecs;
    }

    /**
     * @return lock key
     */
    public String getLockKey() {
        return lockKey;
    }

    private String get(final String key) {
        Object obj = null;
        try {
        	obj=JedisClusterUtil.clusterGetKey(key);
        } catch (Exception e) {
            logger.error("get redis error, key : {}", key);
        }
        return obj != null ? obj.toString() : null;
    }

    private boolean setNX(final String key, final String value) {
        Long result = 0l;
        try {
        	result=JedisClusterUtil.clusterSetNxKey(key, value);
        } catch (Exception e) {
            logger.error("setNX redis error, key : {}", key);
        }
        return result== 1l ? true : false;
    }

    private String getSet(final String key, final String value) {
        String result = null;
        try {
        	result=JedisClusterUtil.clusterGetSetKey(key, value);
        } catch (Exception e) {
            logger.error("setNX redis error, key : {}", key);
        }
        return result;
    }

    /**
     * 得到 lock.
     * 實現思路: 主要是使用了redis 的setnx命令,緩存了鎖.
     * reids緩存的key是鎖的key,全部的共享, value是鎖的到期時間(注意:這裏把過時時間放在value了,沒有時間上設置其超時時間)
     * 執行過程:
     * 1.經過setnx嘗試設置某個key的值,成功(當前沒有這個鎖)則返回,成功得到鎖
     * 2.鎖已經存在則獲取鎖的到期時間,和當前時間比較,超時的話,則設置新的值
     *
     * @return true if lock is acquired, false acquire timeouted
     * @throws InterruptedException in case of thread interruption
     */
    public synchronized boolean lock() throws InterruptedException {
        int timeout = timeoutMsecs;
        while (timeout >= 0) {
            long expires = System.currentTimeMillis() + expireMsecs + 1;
            String expiresStr = String.valueOf(expires); //鎖到期時間
            if (this.setNX(lockKey, expiresStr)) {
                // lock acquired
                locked = true;
                return true;
            }

            String currentValueStr = this.get(lockKey); //redis裏的時間
            if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
                //判斷是否爲空，不爲空的狀況下，若是被其餘線程設置了值，則第二個條件判斷是過不去的
                // lock is expired

                String oldValueStr = this.getSet(lockKey, expiresStr);
                //獲取上一個鎖到期時間，並設置如今的鎖到期時間，
                //只有一個線程才能獲取上一個線上的設置時間，由於jedis.getSet是同步的
                if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                    //防止誤刪（覆蓋，由於key是相同的）了他人的鎖——這裏達不到效果，這裏值會被覆蓋，可是由於什麼相差了不多的時間，因此能夠接受

                    //[分佈式的狀況下]:如過這個時候，多個線程剛好都到了這裏，可是隻有一個線程的設置值和當前值相同，他纔有權利獲取鎖
                    // lock acquired
                    locked = true;
                    return true;
                }
            }
            timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;

            /*
                延遲100 毫秒,  這裏使用隨機時間可能會好一點,能夠防止飢餓進程的出現,即,當同時到達多個進程,
                只會有一個進程得到鎖,其餘的都用一樣的頻率進行嘗試,後面有來了一些進行,也以一樣的頻率申請鎖,這將可能致使前面來的鎖得不到知足.
                使用隨機的等待時間能夠必定程度上保證公平性
             */
            Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);

        }
        return false;
    }


    /**
     * Acqurired lock release.
     */
    public synchronized void unlock() {
        if (locked) {
        	JedisClusterUtil.clusterDelKey(lockKey);
            locked = false;
        }
    }

}

小弟請教了羣裏的大牛，發現這種作法很常見。但一些細節性的問題仍是要好好琢磨一下，他是怎麼實現分佈式鎖的呢？

簡單說 就是講鎖的類型 與 超時時間組合成key-value模式 存放setnx  到redis中。

當多臺服務器 的同一個接口產生併發時，業務正常的狀況下：

c0設置了鎖

c1

if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis())

線程沒法經過這個判斷（c0 的鎖沒過時是大於當前時間的，由於c0的鎖是當前時間加上失效時間的+1的和）。再來個c2也沒法進入。因此正常業務  就是ok的。

當c0線程由於服務宕機或者業務流程過長致使超時呢？ 沒有釋放鎖的時候呢。

上面的if判斷就不能阻擋了，但

if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {

能夠阻擋，當c1和c2併發進入時：

C1使用getSet方法 獲取到c0的失效時間value

C2也執行了getSet方法（失敗的，獲取到的oldvalue就是空串），這就保證了c1的oldValueStr 與currentValueStr是相等 且不爲空，繼而c1獲取到鎖。擁有執行權限，而c2  oldValueStr 與currentValueStr不等（獲取到的是c1的失效時間）。只能繼續循環獲取或者退出

注意：這裏可能致使超時時間不是其本來的超時時間，C1的超時時間可能被C2覆蓋了，可是他們相差的毫秒及其小，這裏忽略了。

可是：釋放鎖 還須要一些注意的地方。那就是判斷一下是否超時

//爲了讓分佈式鎖的算法更穩鍵些，持有鎖的客戶端在解鎖以前應該再檢查一次本身的鎖是否已經超時，再去作DEL操做，由於可能客戶端由於某個耗時的操做而掛起，
            //操做完的時候鎖由於超時已經被別人得到，這時就沒必要解鎖了。

RedisLock redisLock = new RedisLock("userInitialFix");
			try {
				if (redisLock.lock()) {// 獲取鎖,若是成功進行查詢數據庫匹配債權
					selectMaxMoneyByAnyThing = debtOriginalAssetBillsService.selectMaxMoneyByAnyThing(days[i],
							"OLD_PRODUCT", amount);
					DebtOriginalAssetBillsWithBLOBs new_DebtOriginalAsset = new DebtOriginalAssetBillsWithBLOBs();
					if (selectMaxMoneyByAnyThing != null) {
						new_DebtOriginalAsset.setId(selectMaxMoneyByAnyThing.getId());

						new_DebtOriginalAsset.setRemainAmount(selectMaxMoneyByAnyThing.getRemainAmount() - amount);
						new_DebtOriginalAsset.setArrivalAmount(selectMaxMoneyByAnyThing.getArrivalAmount() + amount);
						debtOriginalAssetBillsService.updateSelectiveById(new_DebtOriginalAsset);// 更新債權表
						break;
					}
				} else {
					// 等一秒繼續進行匹配防止無謂循環
					Thread.sleep(1000);
					// 繼續去調用
					return matchDebtOriginalAsset(day, amount);
				}
			} catch (Exception e) {
				System.out.println("用戶初始化按期金額出錯!!");
				e.printStackTrace();
			} finally {
				redisLock.unlock();// 釋放鎖
			}

上述代碼 就差了一個超時的處理。

參考https://www.cnblogs.com/0201zcr/p/5942748.html