其實Redis分布式鎖的介紹钠绍，前面幾篇文章中都要介紹到，只是沒有獨立成篇花沉，今天把其單獨摘出來柳爽，便于學(xué)習(xí)和使用。

1碱屁、概述

當(dāng)多個進(jìn)程不在同一個系統(tǒng)中時磷脯，用分布式鎖控制多個進(jìn)程對資源的操作或者訪問。

分布式鎖的實現(xiàn)要保證幾個基本點：

1娩脾、互斥性：任意時刻赵誓，只有一個資源能夠獲取到鎖
2、容災(zāi)性：能夠在未成功釋放鎖的情況下晦雨，一定時限內(nèi)能夠恢復(fù)鎖的正常功能
3架曹、統(tǒng)一性：加鎖和解鎖保證同一資源來進(jìn)行操作

分布式鎖的實現(xiàn)方式有很多種：

1、數(shù)據(jù)庫樂觀鎖方式（數(shù)據(jù)庫加一個版本號）
2闹瞧、基于Redis的分布式鎖
3绑雄、基于ZK的分布式鎖（Zookeeper基礎(chǔ)（五）：分布式鎖）

2、Redis單機(jī)實現(xiàn)

2.1 原理

https://mp.weixin.qq.com/s?__biz=MzU0OTk3ODQ3Ng==&mid=2247483893&idx=1&sn=32e7051116ab60e41f72e6c6e29876d9&chksm=fba6e9f6ccd160e0c9fa2ce4ea1051891482a95b1483a63d89d71b15b33afcdc1f2bec17c03c&mpshare=1&scene=1&srcid=0416Kx8ryElbpy4xfrPkSSdB&key=1eff032c36dd9b3716bab5844171cca99a4ea696da85eed0e4b2b7ea5c39a665110b82b4c975d2fd65c396e91f4c7b3e8590c2573c6b8925de0df7daa886be53d793e7f06b2c146270f7c0a5963dd26a&ascene=1&uin=MTg2ODMyMTYxNQ%3D%3D&devicetype=Windows+10&version=62060739&lang=zh_CN&pass_ticket=y1D2AijXbuJ8HCPhyIi0qPdkT0TXqKFYo%2FmW07fgvW%2FXxWFJiJjhjTsnInShv0ap

Redisson底層原理簡單描述：
先判斷一個key存在不存在奥邮，如果不存在万牺，則set key，同時設(shè)置過期時間和value（1）洽腺，
這個過程使用lua腳本來實現(xiàn)脚粟，可以保證多個命令的原子性，當(dāng)業(yè)務(wù)完成以后蘸朋，刪除key核无；
如果存在說明已經(jīng)有別的線程獲取鎖了，那么就循環(huán)等待一段時間后再去獲取鎖

如果是可重入鎖呢：
先判斷一個key存在不存在藕坯，如果不存在团南，則set key，同時設(shè)置過期時間和value（線程id:1）炼彪，
如果存在吐根，則判斷value中的線程id是否是當(dāng)前線程的id，如果是辐马，說明是可重入鎖拷橘，則value+1，變成（線程id:2），如果不是冗疮，說明是別的線程來獲取鎖萄唇，則獲取失敗赌厅；這個過程同樣使用lua腳本一次性提交穷绵，保證原子性。

如何防止業(yè)務(wù)還沒執(zhí)行完特愿，但是鎖key過期呢仲墨，可以在線程加鎖成功后，啟動一個后臺進(jìn)程看門狗揍障，去定時檢查目养，如果線程還持有鎖，就延長key的生存時間——Redisson就是這樣實現(xiàn)的毒嫡。

其實Jedis也有現(xiàn)成的實現(xiàn)方式癌蚁，單機(jī)、集群兜畸、分片都有實現(xiàn)努释，底層原理是利用連用setnx、setex指令
（Redis從2.6之后支持setnx咬摇、setex連用）伐蒂，核心是設(shè)置value和設(shè)置過期時間包裝成一個原子操作

jedis.set(key, value, "NX", "PX", expire)

image.png

注：setnx和setex都是原子性的
SETNX key value：
將 key 的值設(shè)為 value ，當(dāng)且僅當(dāng) key 不存在肛鹏；若給定的 key 已經(jīng)存在逸邦，則 SETNX 不做任何動作。
相當(dāng)于是 EXISTS 在扰、SET 兩個命令連用
SETEX key seconds value：
將value關(guān)聯(lián)到key, 并將key的生存時間設(shè)為seconds(以秒為單位)缕减；如果key 已經(jīng)存在，SETEX將重寫舊值芒珠；
相當(dāng)于是SET桥狡、EXPIRE兩個命令連用

2.1 實現(xiàn)

public class RedisTool {

    private static final String LOCK_SUCCESS = "OK";
    //NX|XX, NX -- Only set the key if it does not already exist;
   //        XX -- Only set the key if it already exist.
    private static final String SET_IF_NOT_EXIST = "NX";
    //EX|PX, expire time units: EX = seconds; PX = milliseconds
    private static final String SET_WITH_EXPIRE_TIME = "PX";

   private static volatile JedisPool jedisPool = null;

    public static JedisPool getRedisPoolUtil() {
        if(null == jedisPool ){
            synchronized (RedisTool.class){
                if(null == jedisPool){
                    GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
                    poolConfig.setMaxTotal(100);
                    poolConfig.setMaxIdle(10);
                    poolConfig.setMaxWaitMillis(100*1000);
                    poolConfig.setTestOnBorrow(true);
                    jedisPool = new JedisPool(poolConfig,"192.168.10.151",6379);
                }
            }
        }
        return jedisPool;
    }



    /**
     * 嘗試獲取分布式鎖
     * @param lockKey 鎖
     * @param requestId 請求標(biāo)識
     * @param expireTime 超期時間
     * @return 是否獲取成功
     */
    public static boolean tryGetDistributedLock(String lockKey, String requestId, int expireTime) {
        Jedis  jedis = jedisPool.getResource();

        try {
            String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);

            if (LOCK_SUCCESS.equals(result)) {
                return true;
            }
            return false;
        }catch (Exception e){
            return false;
        }finally {
            jedisPool.returnResource(jedis);
        }

    }

    private static final Long RELEASE_SUCCESS = 1L;

    /**
     * 釋放分布式鎖
     * @param lockKey 鎖
     * @param requestId 請求標(biāo)識
     * @return 是否釋放成功
     */
    public static boolean releaseDistributedLock(String lockKey, String requestId) {

        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        //如果使用的是切片shardedJedis，那么需要先獲取到j(luò)edis皱卓，
        //Jedis jedis = shardedJedis.getShard(key);
        Jedis  jedis = jedisPool.getResource();

        try {
            Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
            if (RELEASE_SUCCESS.equals(result)) {
                return true;
            }
            return false;
        }catch (Exception e){
            return false;
        }finally {
            jedisPool.returnResource(jedis);
        }
    }
}

從jedis源碼中可以發(fā)現(xiàn)上面的加鎖/釋放鎖指令在單機(jī)jedis/ShardedJedis/JedisCluster下都能實現(xiàn)(jedis版本為3.0以上)总放，但是ShardedJedis可以直接加鎖，但是不能直接釋放鎖（沒有提供eval工具方法）好爬，但是可以先
Jedis jedis = shardedJedis.getShard(key) 獲得jedis，然后使用jedis.evel()來釋放鎖甥啄。

注：關(guān)于redisTool工具類的更優(yōu)化實現(xiàn)見Java 函數(shù)式接口編程實例

3 存炮、Cluster集群實現(xiàn)

上面介紹的分布式鎖的實現(xiàn)在Redis Cluster集群模式下，是存在問題的，Redis Cluster集群模式介紹見Redis（四）：集群模式

整個過程如下：

客戶端1在Redis的節(jié)點A上拿到了鎖穆桂；
節(jié)點A宕機(jī)后宫盔，客戶端2發(fā)起獲取鎖key的請求，這時請求就會落在節(jié)點B上享完；
節(jié)點B由于之前并沒有存儲鎖key灼芭，所以客戶端2也可以成功獲取鎖，即客戶端1和客戶端2同時持有了同一個資源的鎖般又。

針對這個問題彼绷。Redis作者antirez提出了RedLock算法來解決這個問題

3.1 RedLock算法

RedLock算法思路如下：

獲取當(dāng)前時間的毫秒數(shù)startTime；
按順序依次向N個Redis節(jié)點執(zhí)行獲取鎖的操作茴迁，這個獲取鎖的操作和前面單Redis節(jié)點獲取鎖的過程相同寄悯，同時鎖超時時間應(yīng)該遠(yuǎn)小于鎖的過期時間；
如果客戶端向某個Redis節(jié)點獲取鎖失敗/超時后堕义，應(yīng)立即嘗試下一個Redis節(jié)點猜旬；
失敗包括Redis節(jié)點不可用或者該Redis節(jié)點上的鎖已經(jīng)被其他客戶端持有
如果客戶端成功獲取到超過半數(shù)的鎖時，記錄當(dāng)前時間endTime倦卖，同時計算整個獲取鎖過程的總耗時costTime = endTime - startTime洒擦，如果獲取鎖總共消耗的時間遠(yuǎn)小于鎖的過期時間（即costTime < expireTime），則認(rèn)為客戶端獲取鎖成功怕膛，否則熟嫩，認(rèn)為獲取鎖失敗
如果獲取鎖成功，需要重新計算鎖的過期時間嘉竟。它等于最初鎖的有效時間減去第三步計算出來獲取鎖消耗的時間邦危，即expireTime - costTime
如果最終獲取鎖失敗，那么客戶端立即向所有Redis發(fā)起釋放鎖的操作舍扰。（和單機(jī)釋放鎖的邏輯一樣）

3.2 缺陷

RedLock算法雖然可以解決單點Redis分布式鎖的安全性問題倦蚪，但如果集群中有節(jié)點發(fā)生崩潰重啟，還是會對鎖的安全性有影響的边苹。

假設(shè)一共有5個Redis節(jié)點：A, B, C, D, E陵且。設(shè)想發(fā)生了如下的事件序列：

客戶端1成功鎖住了A, B, C，獲取鎖成功（但D和E沒有鎖赘鍪）慕购；
節(jié)點C崩潰重啟了，但客戶端1在C上加的鎖沒有持久化下來茬底，丟失了沪悲；
節(jié)點C重啟后，客戶端2鎖住了C, D, E阱表，獲取鎖成功殿如；

這樣贡珊，客戶端1和客戶端2同時獲得了鎖（針對同一資源）。針對這樣場景涉馁，解決方式也很簡單门岔，也就是讓Redis崩潰后延遲重啟，并且這個延遲時間大于鎖的過期時間就好烤送。這樣等節(jié)點重啟后寒随，所有節(jié)點上的鎖都已經(jīng)失效了。也不存在以上出現(xiàn)2個客戶端獲取同一個資源的情況了

還有一種情況帮坚，如果客戶端1獲取鎖后妻往，訪問共享資源操作執(zhí)行任務(wù)時間過長（要么邏輯問題，要么發(fā)生了GC）叶沛，導(dǎo)致鎖過期了蒲讯，而后續(xù)客戶端2獲取鎖成功了，這樣就會導(dǎo)致客戶端1和客戶端2同時操作共享資源灰署，相當(dāng)于同一個時刻出現(xiàn)了2個客戶端獲得了鎖的情況判帮。這也就是上面鎖過期時間要遠(yuǎn)遠(yuǎn)大于加鎖消耗的時間的原因。
服務(wù)器臺數(shù)越多溉箕，出現(xiàn)不可預(yù)期的情況也越多晦墙，所以針對分布式鎖的應(yīng)用的時候需要多測試。
如果系統(tǒng)對共享資源有非常嚴(yán)格要求得情況下肴茄，還是建議需要做數(shù)據(jù)庫鎖的方案來補(bǔ)充晌畅，如飛機(jī)票或火車票座位得情況。
對于一些搶購獲取寡痰，針對偶爾出現(xiàn)超賣抗楔，后續(xù)可以通過人工介入來處理，畢竟redis節(jié)點不是天天奔潰拦坠，同時數(shù)據(jù)庫鎖的方案
性能又低连躏。

3.3 實現(xiàn)

redisson包已經(jīng)有對redlock算法封裝

public interface DistributedLock {
    /**
     * 獲取鎖
     * @author zhi.li
     * @return 鎖標(biāo)識
     */
    String acquire();

    /**
     * 釋放鎖
     * @author zhi.li
     * @param indentifier
     * @return
     */
    boolean release(String indentifier);
}

public class RedisDistributedRedLock implements DistributedLock {

    /**
     * redis 客戶端
     */
    private RedissonClient redissonClient;

    /**
     * 分布式鎖的鍵值
     */
    private String lockKey;

    private RLock redLock;

    /**
     * 鎖的有效時間 10s
     */
    int expireTime = 10 * 1000;

    /**
     * 獲取鎖的超時時間
     */
    int acquireTimeout  = 500;

    public RedisDistributedRedLock(RedissonClient redissonClient, String lockKey) {
        this.redissonClient = redissonClient;
        this.lockKey = lockKey;
    }

    @Override
    public String acquire() {
        redLock = redissonClient.getLock(lockKey);
        boolean isLock;
        try{
            isLock = redLock.tryLock(acquireTimeout, expireTime, TimeUnit.MILLISECONDS);
            if(isLock){
                System.out.println(Thread.currentThread().getName() + " " + lockKey + "獲得了鎖");
                return null;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public boolean release(String indentifier) {
        if(null != redLock){
            redLock.unlock();
            return true;
        }

        return false;
    }
}

4、項目中調(diào)用

RedisTool 中加鎖/釋放鎖實現(xiàn)后贞滨，在項目中怎么調(diào)用呢入热，如果直接在業(yè)務(wù)代碼中調(diào)用，那一方面太麻煩了晓铆，另一方面耦合太多勺良，如果有一天需要改動其中的邏輯，那在項目中需要改動很多地方骄噪。

這里我們使用AOP+注解來實現(xiàn)調(diào)用尚困，即在需要加鎖的方法上添加注解，然后再AOP中链蕊，統(tǒng)一加鎖尾组，釋放鎖忙芒。

4.1 自定義注解

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisLockAnnotation {
    int expire() default 5;

    String field() default "";

}

4.2 自定義切面

@Aspect
@Service
public class RedisLockAspect {

      //方法切點
     @Pointcut("@annotation(redisLock.RedisLockAnnotation)")
     public  void methodAspect() {

     }

    @Around("methodAspect()")
    public Object execute(ProceedingJoinPoint joinPoint) throws Throwable {
        Signature signature = joinPoint.getSignature();
        Method method = ((MethodSignature) signature).getMethod();
        Method realMethod = joinPoint.getTarget().getClass().getDeclaredMethod(signature.getName(),method.getParameterTypes());
        RedisLockAnnotation redisLockAnnotation = realMethod.getAnnotation(RedisLockAnnotation.class);
        int expireTime = redisLockAnnotation.expire();
        String field = redisLockAnnotation.field();

        Map<String, Object> params = getNameAndValue(joinPoint, field);
        if (params==null){
            throw new RuntimeException("params is not allowed null");
        }
        String url = method.getDeclaringClass().getSimpleName() + "." + method.getName();
        String reqParam = JSONObject.toJSONString(params);

        //redis加鎖
        String localKey = url + ":" + reqParam;
        String requestFlag = UUID.randomUUID().toString();
        boolean lock = RedisTool.tryGetDistributedLock(localKey, requestFlag, expireTime);
        if(!lock){
            return "鎖已存在";
        }

        //加鎖成功
        Object result = null;
        try {
            //執(zhí)行方法
            result =joinPoint.proceed();
        } finally {
            //方法執(zhí)行完之后進(jìn)行解鎖
            RedisTool.releaseDistributedLock(localKey, requestFlag);
        }

        return result;
    }


    /**
     * 獲取參數(shù)Map集合
     */
    private Map<String, Object> getNameAndValue(ProceedingJoinPoint joinPoint, String filedList) {
        Map<String, Object> param = new HashMap<String, Object>();
        Object[] paramValues = joinPoint.getArgs();
        String[] paramNames = ((CodeSignature) joinPoint.getSignature()).getParameterNames();
        for (int i = 0; i < paramValues.length; i++) {
            List<String> targetFields = Arrays.asList(filedList.split(","));
            JSONObject valueDetialsJson = (JSONObject) JSONObject.toJSON(paramValues[i]);
            //得到屬性
            for (int j = 0; j < targetFields.size(); j++) {
                if (valueDetialsJson.get(targetFields.get(i))!=null){
                    param.put(targetFields.get(i), valueDetialsJson.get(targetFields.get(i)));
                }
            }

        }
        if (param != null && param.size() > 0) {
            return param;
        }
        for (int i = 0; i < paramNames.length; i++) {
            param.put(paramNames[i], paramValues[i]);
        }

        return param;
    }

}

4.3 使用

public class RedidLockTest1 {

    @RedisLockAnnotation(field = "userId")
    public Object test1(String userId){
        return userId+"==";
    }

}

Redis（八）：Redis分布式鎖實現(xiàn)