經過線上實戰的redis 分散式鎖程式碼。

    能用，但是效能較差。

已考慮:

    1.只能被擁有鎖的執行緒解鎖

    2. 設定節點和超時時間用同一個key

 未考慮:

 1. 不能重入

 2. 沒有本地鎖，併發效能會比較差，不使用用在併發爭鎖較多的場景下。本地鎖非自旋

 3. 未考慮鎖等待排序. 這個是redis很難實現的.

     可以通過redis的list實現,但缺點是list下每個子節點無超時時間. redis也無法進行模糊查詢 key*.

     故還是通過zookeeper實現比較好. 但zookeeper 會遇到效能瓶頸,我們線下的就出現了,經常註冊不上的情況.

    zookeeper原理是臨時節點

使用方式：

GlobalLockRedisImpl globalLockRedis = new GlobalLockRedisImpl(casRedis, maxLockSeconds, sleepTimeMillis,
                redisKey);
            //獲取分散式加鎖
            globalLockRedis.lock();

try{

   // do something

}finnaly{

 // 分散式鎖釋放
                globalLockRedis.unlock();
}

    private boolean setIfAbsent(String key, String value, int expireMilliSeconds) {
            String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds);
            if (result != null && result.equalsIgnoreCase("OK")) {
                return true;
            }
            return false;

   

程式碼：

/**
 * 基於redis實現的全域性鎖,不能當做單例使用.
 *
 * @author loufei
 *
 *         2015-5-28
 */
public class GlobalLockRedisImpl implements GlobalLock {
    private static ILog           LOGGER = LogFactory.getLog(GlobalLockRedisImpl.class);

    private final CasRedisService casRedis;
    private final int             maxLockSeconds;
    private final long            sleepTimeMillis;
    private Thread                exclusiveOwnerThread;
    private final String          key;

    public GlobalLockRedisImpl(CasRedisService casRedis, int maxLockSeconds, long sleepTimeMillis, String key) {
        this.casRedis = casRedis;
        this.maxLockSeconds = maxLockSeconds;
        if (maxLockSeconds > 30) {
            maxLockSeconds = 30;
        }
        this.sleepTimeMillis = sleepTimeMillis;
        if (sleepTimeMillis > 1000) {
            sleepTimeMillis = 1000;
        }
        this.key = key;
    }

    @Override
    public void lock() {
        long startTime = System.currentTimeMillis();
        int tryCount = 0;
        while (true) {
            tryCount++;
            // setIfAbsent tps對於併發鎖控制夠用了.
            Boolean result = casRedis.setIfAbsent(key, "1", maxLockSeconds * 1000);
            // 處理鎖的自動釋放,前三次嘗試加鎖都會進行超時設定,保證分散式鎖有timeOut.避免主加鎖程序被無故停止,導致key無失效時間,鎖永遠不被釋放.

            if (result == null || !result) {// 加鎖失敗,阻塞呼叫執行緒
                LOGGER.info(" spin ,key=" + key + ",tryCount=" + tryCount);
                try {
                    Thread.sleep(sleepTimeMillis);
                } catch (InterruptedException e) {
                    // nothing need to be done
                }
                long end = System.currentTimeMillis();

                //超時
                long costTime = end - startTime;
                // 分散式自旋等待時間已經已經超過了某個時間(暫定位為3秒),說明分散式競爭失敗或者key沒有正確的被設定超時時間.
                long sleepMillisWaterMark = TimeUnit.SECONDS.toMillis(2);
                if (costTime > sleepMillisWaterMark) {
                    String message = "get redis global lock error .1. compete failed 2. key do not set  timeOut ,exist for ever  ,maxLockSeconds="
                                     + maxLockSeconds
                                     + ",costTimeMillis="
                                     + costTime
                                     + ",key="
                                     + key
                                     + ",retryCount="
                                     + tryCount
                                     + ",sleepTimeMillis="
                                     + sleepTimeMillis
                                     + ",costTimePerTime="
                                     + (costTime / ((double) tryCount));
                    // 拋執行期異常,冪等重複執行不會總是拋錯.
                    throw new GlobalLockTimeOutException(message);
                }
                continue;
            }
            exclusiveOwnerThread = Thread.currentThread();

            break;
        }
        long endTime = System.currentTimeMillis();
        long costTime = endTime - startTime;
        LOGGER.info("get redis global lock success,maxLockSeconds=" + maxLockSeconds + ",costTimeMillis=" + costTime
                    + ",key=" + key + ",retryCount=" + tryCount + ",sleepTimeMillis=" + sleepTimeMillis
                    + ",costTimePerTime=" + costTime / ((double) tryCount) + ",startTime=" + startTime + ",endTime="
                    + endTime);
    }

    @Override
    public void unlock() {
        long startTime = System.currentTimeMillis();
        if (exclusiveOwnerThread == Thread.currentThread()) {
            Integer del = casRedis.del(key.getBytes());
            long endTime = System.currentTimeMillis();

            exclusiveOwnerThread = null;
            LOGGER.info(" global unlock,del count=" + del + ".key=" + key + " ,costTime millis="
                        + (endTime - startTime) + ",startTime=" + startTime + ",endTime=" + endTime);
        } else {
            LOGGER.info(" thread do not get lock ,can not unlock. key=" + key + ",exclusiveOwnerThread="
                        + exclusiveOwnerThread + ",current thrad=" + Thread.currentThread());
        }

    }

    public int getMaxLockSeconds() {
        return maxLockSeconds;
    }
}

    private boolean setIfAbsent(String key, String value, int expireMilliSeconds) {
            String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds);
            if (result != null && result.equalsIgnoreCase("OK")) {
                return true;
            }
            return false;

   }

版本2 , 未經線上驗證.

/**
 * 基於redis實現的全域性鎖,不能當做單例使用.
 * 
 * @author loufei
 * 
 *         2015-5-28
 */
public class GlobalLockSeqRedisImpl implements GlobalLock {
    private static long maxWaitSeconds = 2;
    private static final String split = "____";
    private static AtomicInteger seqCount = new AtomicInteger();

    private static String host = null;
    private static ILog LOGGER = LogFactory.getLog(GlobalLockRedisImpl.class);

    private final static ExecutorService threadPoolExecutor = TaxiExecutors
            .newCachedThreadPool(new ThreadFactoryBuilder("GlobalLockRedisImpl"));
    private final didikuaidi.redis.clients.jedis.JedisCommands casRedis;
    private final int maxLockSeconds;
    private final long sleepTimeMillis;
    private Thread exclusiveOwnerThread;
    private final String key;
    private final Lock lock = new ReentrantLock();
    private String value = null;

    static {
        InetAddress localHost = null;
        try {
            localHost = InetAddress.getLocalHost();
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
        host = (UUID.randomUUID() + localHost.getHostAddress()).replace(split, "");
    }

    public GlobalLockSeqRedisImpl(JedisCommands casRedis, int maxLockSeconds, long sleepTimeMillis, String key) {
        this.casRedis = casRedis;
        this.maxLockSeconds = maxLockSeconds;
        if (maxLockSeconds > 30) {
            maxLockSeconds = 30;
        }
        this.sleepTimeMillis = sleepTimeMillis;
        if (sleepTimeMillis > 1000) {
            sleepTimeMillis = 1000;
        }
        this.key = key;
    }

    private boolean setIfAbsent(String key, String value, int expireMilliSeconds) {
        String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds);
        if (result != null && result.equalsIgnoreCase("OK")) {
            return true;
        }
        return false;

    }

    private int lockRedisSeq(long startMillis) {
        int seqNo = seqCount.incrementAndGet();

        value = host + split + new Date().getTime() + split + seqNo;
        Long listSize = casRedis.lpush(key, value);
        int count = 1;
        LOGGER.info("listSize=" + listSize);
        if (listSize.longValue() == 1l) {
            // 長度是1,說明只有自己拿到鎖.
            return count;
        } else {
            // 如果鎖過多,列印error報警
            if (listSize > DynamicConfig.getInt(DynamicConfigKeys.KUAIPAY_LOCK_LIMIT, 21)) {
                LOGGER.error("too_much_lock_node listSize=" + listSize);
            } else {
                LOGGER.info("lock_queue_size=" + listSize + ",key=" + key);
            }
            // 說明沒有鎖住
            while (true) {
                try {
                    Thread.sleep(sleepTimeMillis);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // -1代表從尾部獲取.FIFO原則
                String tailNodeValue = casRedis.lindex(key, -1);
                if (tailNodeValue == null) {
                    // 自己的節點莫名其妙被人刪除了
                    String message = "tailNodeValue is null ,all node is delete inclued itself key=" + key;
                    LOGGER.error(message);
                    throw new GlobalLockTimeOutException(message);
                } else if (tailNodeValue.equals(value)) {
                    // 拿到鎖了
                    return count;
                } else {
                    String[] split = tailNodeValue.split(GlobalLockSeqRedisImpl.split);
                    String host = split[0];
                    // 保持資料相容性避免報錯
                    Date nodeCreateTime = new Date(Long.valueOf(split[1]));
                    long diffInMillis = new Date().getTime() - nodeCreateTime.getTime();
                    long maxLockTime = TimeUnit.MINUTES.toMillis(1);
                    if (host.equals(this.host) && diffInMillis > TimeUnit.SECONDS.toMillis(3)) {
                        // 發現當前節點的機器是本機(不用擔心不同機器的時間不一致問題),且超過3秒,立即刪除.
                        lrem(tailNodeValue);
                    }
                    if (diffInMillis > TimeUnit.SECONDS.toMillis(30) && diffInMillis <= maxLockTime) {
                        // 大量error報警,發現問題.請檢查對應ip機器和日誌所在機器的時間是否一致.並且java預設時區是否一致
                        LOGGER.error("Lock Node donot unLock ,請檢查對應機器的時間是否一致,key=" + key + " unlockNodeIp= " + split[0]
                                + ",nodeCreateTime=" + nodeCreateTime);
                    } else if (diffInMillis > maxLockTime) {
                        // 判斷下別人節點的時間是否超時
                        LOGGER.error("Lock Node donot unLock ,請檢查對應機器的時間是否一致,key=" + key + " unlockNodeIp= " + split[0]
                                + ",nodeCreateTime=" + nodeCreateTime);
                        lrem(tailNodeValue);

                    }
                }

                long now = System.currentTimeMillis();
                long costTime = now - startMillis;
                count++;

                if (costTime > TimeUnit.SECONDS.toMillis(maxLockSeconds)) {
                    String errorString = getErrorString(count, costTime);
                    throw new GlobalLockTimeOutException(errorString);
                }

            }
        }
        //
    }

    private void lrem(String tailNodeValue) {
        // -1代表從尾部獲取.FIFO原則.採用rem時間複雜度可能增加,這種情況畢竟少見
        long remCount = casRedis.lrem(key, -2, tailNodeValue);
        if (remCount != 1) {
            LOGGER.error("del count dot not  euqal 1,remCount=" + remCount);
        }
    }

    @Override
    public void lock() {
        long startTime = System.currentTimeMillis();

        // 搶佔本地鎖,會自動喚醒
        try {
            boolean success = lock.tryLock(maxWaitSeconds, TimeUnit.SECONDS);
            if (!success) {
                long end = System.currentTimeMillis();
                // 超時
                long costTime = end - startTime;
                String message = getErrorString(1, costTime);
                LOGGER.error("lock timeOut " + message);
                throw new GlobalLockTimeOutException(message);
            }
        } catch (Exception e) {
            long end = System.currentTimeMillis();
            // 被其他人中斷
            long costTime = end - startTime;
            String message = getErrorString(1, costTime);
            LOGGER.error("lock meet exception " + message, e);
            throw new GlobalLockTimeOutException(message);
        }
        int tryCount = lockRedisSeq(startTime);
        exclusiveOwnerThread = Thread.currentThread();
        long endTime = System.currentTimeMillis();
        long costTime = endTime - startTime;
        LOGGER.info("get redis global lock success,maxLockSeconds=" + maxLockSeconds + ",costTimeMillis=" + costTime
                + ",key=" + key + ",retryCount=" + tryCount + ",sleepTimeMillis=" + sleepTimeMillis
                + ",costTimePerTime=" + costTime / ((double) tryCount) + ",startTime=" + startTime + ",endTime="
                + endTime);
    }

    private int lockRedis(long startTime) {
        // 搶佔遠端分散式鎖,無解鎖通知,故採用自旋等待
        int tryCount = 0;
        while (true) {
            tryCount++;
            // setIfAbsent tps對於併發鎖控制夠用了.
            Boolean result = this.setIfAbsent(key, "1", maxLockSeconds * 1000);
            // 處理鎖的自動釋放,前三次嘗試加鎖都會進行超時設定,保證分散式鎖有timeOut.避免主加鎖程序被無故停止,導致key無失效時間,鎖永遠不被釋放.

            if (result == null || !result) {// 加鎖失敗,阻塞呼叫執行緒
                LOGGER.info(" spin ,key=" + key + ",tryCount=" + tryCount);
                try {
                    Thread.sleep(sleepTimeMillis);
                } catch (InterruptedException e) {
                    // nothing need to be done
                }
                long end = System.currentTimeMillis();

                // 超時
                long costTime = end - startTime;
                // 分散式自旋等待時間已經已經超過了某個時間(暫定位為3秒),說明分散式競爭失敗或者key沒有正確的被設定超時時間.
                long sleepMillisWaterMark = TimeUnit.SECONDS.toMillis(maxWaitSeconds);
                if (costTime > sleepMillisWaterMark) {
                    String message = getErrorString(tryCount, costTime);
                    // 拋執行期異常,冪等重複執行不會總是拋錯.
                    throw new GlobalLockTimeOutException(message);
                }
                continue;
            }
            exclusiveOwnerThread = Thread.currentThread();

            break;
        }
        return tryCount;
    }

    private String getErrorString(int tryCount, long costTime) {
        return "get redis global lock error .1. compete failed 2. key do not set  timeOut ,exist for ever  ,maxLockSeconds="
                + maxLockSeconds + ",costTimeMillis=" + costTime + ",key=" + key + ",retryCount=" + tryCount
                + ",sleepTimeMillis=" + sleepTimeMillis + ",costTimePerTime=" + (costTime / ((double) tryCount));
    }

    @Override
    public void unlock() {
        long startTime = System.currentTimeMillis();
        try {
            lock.unlock();
        } catch (IllegalMonitorStateException e) {
            LOGGER.error("IllegalMonitorStateException", e);
        }
        if (exclusiveOwnerThread == Thread.currentThread()) {

            // 解鎖,如果解鎖失敗,重試解鎖三次
            Long delCount = unlockSeqRedisAndRetryIfError();
            long endTime = System.currentTimeMillis();
            exclusiveOwnerThread = null;
            LOGGER.info(" global unlock,del count=" + delCount + ".key=" + key + " ,costTime millis="
                    + (endTime - startTime) + ",startTime=" + startTime + ",endTime=" + endTime);
        } else {
            LOGGER.info(" thread do not get lock ,can not unlock. key=" + key + ",exclusiveOwnerThread="
                    + exclusiveOwnerThread + ",current thrad=" + Thread.currentThread());
        }

    }

    private Long unlockSeqRedisAndRetryIfError() {
        Long delCount = 0l;
        try {
            delCount = unlockSeqRedisLock();
        } catch (Exception e) {
            LOGGER.error("unlockSeqRedisLock error", e);
            threadPoolExecutor.execute(getRetryUnlockTask());
        }
        return delCount;
    }

    private Runnable getRetryUnlockTask() {
        return new Runnable() {
            @Override
            public void run() {
                sleepSlience();
                for (int i = 0; i < 3; i++) {
                    try {
                        unlockSeqRedisLock();
                    } catch (Exception e) {
                        // 出現網路超時等異常情況時,重試
                        LOGGER.error("unlockSeqRedisLock error", e);
                        sleepSlience();
                        continue;
                    }
                    break;
                }
            }

            private void sleepSlience() {
                try {
                    Thread.sleep(1000l);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        };
    }

    @Deprecated
    // 請使用unlockSeqRedisAndRetryIfError
    private Long unlockSeqRedisLock() {
        // -1代表從尾部刪除.FIFO原則
        String lastNodeValue = casRedis.rpop(key);
        if (!value.equals(lastNodeValue)) {
            LOGGER.error("del result do not match expect value,expect=" + value + ",acutal value=" + lastNodeValue);
            // 不能用Rpush,否則將導致鎖節點被隨意變更,造成混亂.
            casRedis.lpush(key, lastNodeValue);
            Long lrem = casRedis.lrem(key, -2, value);
            if (lrem != 1) {
                LOGGER.error("del resutl do not match expect value,expect" + value);
            }
            return lrem;
        }
        return 1l;

    }

    private Long unlockRedisLock() {
        return casRedis.del(key);
    }

    public int getMaxLockSeconds() {
        return maxLockSeconds;
    }
}