Redis的分布式鎖應該怎么打開

這篇文章主要講解了“Redis的分布式鎖應該怎么打開”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Redis的分布式鎖應該怎么打開”吧！

要求

基于Redis實現分布式鎖需要滿足如下幾點要求：

1. 在分布式集群中，被分布式鎖控制的方法或代碼段同一時刻只能被一個客戶端上面的一個線程執(zhí)行，也就是互斥

2. 鎖信息需要設置過期時間，避免一個線程長期占有（比如在做解鎖操作前異常退出）而導致死鎖

3. 加鎖與解鎖必須一致，誰加的鎖，就由誰來解（或過期超時），一個客戶端不能解開另一個客戶端加的鎖

4. 加鎖與解鎖的過程必須保證原子性

實現

1. 加鎖實現

基于Redis的分布式鎖加鎖操作一般使用 SETNX 命令，其含義是“將 key 的值設為 value ，當且僅當 key 不存在。若給定的 key 已經存在，則 SETNX 不做任何動作”。
在 Spring Boot 中，可以使用 StringRedisTemplate 來實現，如下，一行代碼即可實現加鎖過程。（下列代碼給出兩種調用形式——立即返回加鎖結果與給定超時時間獲取加鎖結果）

/**
    * 嘗試獲取鎖（立即返回）
    * @param key  鎖的redis key
    * @param value 鎖的value
    * @param expire 過期時間/秒
    * @return 是否獲取成功
    */public boolean lock(String key, String value, long expire) {    return stringRedisTemplate.opsForValue().setIfAbsent(key, value, expire, TimeUnit.SECONDS);
}/**
    * 嘗試獲取鎖，并至多等待timeout時長
    *
    * @param key  鎖的redis key
    * @param value 鎖的value
    * @param expire 過期時間/秒
    * @param timeout 超時時長
    * @param unit    時間單位
    * @return 是否獲取成功
    */public boolean lock(String key, String value, long expire, long timeout, TimeUnit unit) {    long waitMillis = unit.toMillis(timeout);    long waitAlready = 0;    while (!stringRedisTemplate.opsForValue().setIfAbsent(key, value, expire, TimeUnit.SECONDS) && waitAlready < waitMillis) {        try {
            Thread.sleep(waitMillisPer);
        } catch (InterruptedException e) {
            log.error("Interrupted when trying to get a lock. key: {}", key, e);
        }
        waitAlready += waitMillisPer;
    }    if (waitAlready < waitMillis) {        return true;
    }
    log.warn("<====== lock {} failed after waiting for {} ms", key, waitAlready);    return false;
}

上述實現如何滿足前面提到的幾點要求：

1. 客戶端互斥： 可以將expire過期時間設置為大于同步代碼的執(zhí)行時間，比如同步代碼塊執(zhí)行時間為1s，則可將expire設置為3s或5s。避免同步代碼執(zhí)行過程中expire時間到，其它客戶端又可以獲取鎖執(zhí)行同步代碼塊。

2. 通過設置過期時間expire來避免某個客戶端長期占有鎖。

3. 通過value來控制誰加的鎖，由誰解的邏輯，比如可以使用requestId作為value，requestId唯一標記一次請求。

4. setIfAbsent方法 底層通過調用 Redis 的 SETNX 命令，操作具備原子性。

錯誤示例：

網上有如下實現，

public boolean lock(String key, String value, long expire) {    boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, value);    if(result) {
        stringRedisTemplate.expire(key, expire, TimeUnit.SECONDS);
    }    return result;
}

該實現的問題是如果在result為true，但還沒成功設置expire時，程序異常退出了，將導致該鎖一直被占用而導致死鎖，不滿足第二點要求。

2. 解鎖實現

解鎖也需要滿足前面所述的四個要求，實現代碼如下：

private static final String RELEASE_LOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";private static final Long RELEASE_LOCK_SUCCESS_RESULT = 1L;/**
    * 釋放鎖
    * @param key  鎖的redis key
    * @param value 鎖的value
    */public boolean unLock(String key, String value) {
    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(RELEASE_LOCK_LUA_SCRIPT, Long.class);    long result = stringRedisTemplate.execute(redisScript, Collections.singletonList(key), value);    return Objects.equals(result, RELEASE_LOCK_SUCCESS_RESULT);
}

這段實現使用一個Lua腳本來實現解鎖操作，保證操作的原子性。傳入的value值需與該線程加鎖時的value一致，可以使用requestId（具體實現下面給出）。

錯誤示例：

   public boolean unLock(String key, String value) {
        String oldValue = stringRedisTemplate.opsForValue().get(key);        if(value.equals(oldValue)) {
            stringRedisTemplate.delete(key);
        }
}

該實現先獲取鎖的當前值，判斷兩值相等則刪除?？紤]一種極端情況，如果在判斷為true時，剛好該鎖過期時間到，另一個客戶端加鎖成功，則接下來的delete將不管三七二十一將別人加的鎖直接刪掉了，不滿足第三點要求。該示例主要是因為沒有保證解鎖操作的原子性導致。

3. 注解支持

為了方便使用，添加一個注解，可以放于方法上控制方法在分布式環(huán)境中的同步執(zhí)行。

/**
* 標注在方法上的分布式鎖注解
*/@Retention(RetentionPolicy.RUNTIME)@Target(ElementType.METHOD)public @interface DistributedLockable {    String key();    String prefix() default "disLock:";    long expire() default 10L; // 默認10s過期}

添加一個切面來解析注解的處理，

/**
* 分布式鎖注解處理切面
*/@Aspect@Slf4jpublic class DistributedLockAspect {    private DistributedLock lock;    public DistributedLockAspect(DistributedLock lock) {        this.lock = lock;
    }    /**
     * 在方法上執(zhí)行同步鎖
     */
    @Around(value = "@annotation(lockable)")    public Object distLock(ProceedingJoinPoint point, DistributedLockable lockable) throws Throwable {        boolean locked = false;
        String key = lockable.prefix() + lockable.key();        try {
            locked = lock.lock(key, WebUtil.getRequestId(), lockable.expire());            if(locked) {                return point.proceed();
            } else {
                log.info("Did not get a lock for key {}", key);                return null;
            }
        } catch (Exception e) {            throw e;
        } finally {            if(locked) {                if(!lock.unLock(key, WebUtil.getRequestId())){
                    log.warn("Unlock {} failed, maybe locked by another client already. ", lockable.key());
                }
            }
        }
    }
}

RequestId 的實現如下，通過注冊一個Filter，在請求開始時生成一個uuid存于ThreadLocal中，在請求返回時清除。

public class WebUtil {    public static final String REQ_ID_HEADER = "Req-Id";    private static final ThreadLocal<String> reqIdThreadLocal = new ThreadLocal<>();    public static void setRequestId(String requestId) {
        reqIdThreadLocal.set(requestId);
    }    public static String getRequestId(){
        String requestId = reqIdThreadLocal.get();        if(requestId == null) {
            requestId = ObjectId.next();
            reqIdThreadLocal.set(requestId);
        }        return requestId;
    }    public static void removeRequestId() {
        reqIdThreadLocal.remove();
    }
}public class RequestIdFilter implements Filter {    @Override
    public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
        HttpServletRequest httpServletRequest = (HttpServletRequest) servletRequest;
        String reqId = httpServletRequest.getHeader(WebUtil.REQ_ID_HEADER);        //沒有則生成一個
        if (StringUtils.isEmpty(reqId)) {
            reqId = ObjectId.next();
        }
        WebUtil.setRequestId(reqId);        try {
            filterChain.doFilter(servletRequest, servletResponse);
        } finally {
            WebUtil.removeRequestId();
        }
    }
}//在配置類中注冊Filter/**
* 添加RequestId
* @return*/@Beanpublic FilterRegistrationBean requestIdFilter() {
    RequestIdFilter reqestIdFilter = new RequestIdFilter();
    FilterRegistrationBean registrationBean = new FilterRegistrationBean();
    registrationBean.setFilter(reqestIdFilter);
    List<String> urlPatterns = Collections.singletonList("/*");
    registrationBean.setUrlPatterns(urlPatterns);
    registrationBean.setOrder(Ordered.HIGHEST_PRECEDENCE + 1);    return registrationBean;
}

4. 使用注解

@DistributedLockable(key = "test", expire = 10)public void test(){
    System.out.println("線程-"+Thread.currentThread().getName()+"開始執(zhí)行..." + LocalDateTime.now());    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("線程-"+Thread.currentThread().getName()+"結束執(zhí)行..." + LocalDateTime.now());
}

總結

本文給出了基于Redis的分布式鎖的實現方案與常見的錯誤示例。要保障分布式鎖的正確運行，需滿足本文所提的四個要求，尤其注意保證加鎖解鎖操作的原子性，設置過期時間，及對同一個鎖的加鎖解鎖線程一致。

感謝各位的閱讀，以上就是“Redis的分布式鎖應該怎么打開”的內容了，經過本文的學習后，相信大家對Redis的分布式鎖應該怎么打開這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！