基於Redis分佈式鎖的正確打開方式

時間 2020-07-17

原文原文鏈接

分佈式鎖是在分佈式環境下（多個JVM進程）控制多個客戶端對某一資源的同步訪問的一種實現，與之相對應的是線程鎖，線程鎖控制的是同一個JVM進程內多個線程之間的同步。分佈式鎖的通常實現方法是在應用服務器以外經過一個共享的存儲服務器存儲鎖資源，同一時刻只有一個客戶端能佔有鎖資源來完成。一般有基於Zookeeper，Redis，或數據庫三種實現形式。本文介紹基於Redis的實現方案。html

要求

基於Redis實現分佈式鎖須要知足以下幾點要求：java

在分佈式集羣中，被分佈式鎖控制的方法或代碼段同一時刻只能被一個客戶端上面的一個線程執行，也就是互斥
鎖信息須要設置過時時間，避免一個線程長期佔有（好比在作解鎖操做前異常退出）而致使死鎖
加鎖與解鎖必須一致，誰加的鎖，就由誰來解（或過時超時），一個客戶端不能解開另外一個客戶端加的鎖
加鎖與解鎖的過程必須保證原子性

實現

1. 加鎖實現

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

/**
    * 嘗試獲取鎖（當即返回）
    * @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;
}複製代碼

上述實現如何知足前面提到的幾點要求：數據庫

客戶端互斥：能夠將expire過時時間設置爲大於同步代碼的執行時間，好比同步代碼塊執行時間爲1s，則可將expire設置爲3s或5s。避免同步代碼執行過程當中expire時間到，其它客戶端又能夠獲取鎖執行同步代碼塊。
經過設置過時時間expire來避免某個客戶端長期佔有鎖。
經過value來控制誰加的鎖，由誰解的邏輯，好比可使用requestId做爲value，requestId惟一標記一次請求。
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時，程序異常退出了，將致使該鎖一直被佔用而致使死鎖，不知足第二點要求。ide

2. 解鎖實現

解鎖也須要知足前面所述的四個要求，實現代碼以下：ui

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（具體實現下面給出）。this

錯誤示例：url

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

該實現先獲取鎖的當前值，判斷兩值相等則刪除。考慮一種極端狀況，若是在判斷爲true時，恰好該鎖過時時間到，另外一個客戶端加鎖成功，則接下來的delete將無論三七二十一將別人加的鎖直接刪掉了，不知足第三點要求。該示例主要是由於沒有保證解鎖操做的原子性致使。

3. 註解支持

爲了方便使用，添加一個註解，能夠放於方法上控制方法在分佈式環境中的同步執行。

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

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

/**
* 分佈式鎖註解處理切面
*/
@Aspect
@Slf4j
public class DistributedLockAspect {

    private DistributedLock lock;

    public DistributedLockAspect(DistributedLock lock) {
        this.lock = lock;
    }

    /**
     * 在方法上執行同步鎖
     */
    @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
*/
@Bean
public 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()+"開始執行..." + LocalDateTime.now());
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("線程-"+Thread.currentThread().getName()+"結束執行..." + LocalDateTime.now());
}複製代碼