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

186次閱讀

沒有評論

共計 6200 個字符，預計需要花費 16 分鐘才能閱讀完成。

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

要求

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

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

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

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

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

實現 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;
}

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

客戶端互斥：可以將 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 時，程序異常退出了，將導致該鎖一直被占用而導致死鎖，不滿足第二點要求。

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. 注解支持

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

/**
*  標注在方法上的分布式鎖注解
*/@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;
 } /**
 *  在方法上執行同步鎖
 */
 @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()+ 開始執行...  + LocalDateTime.now()); try { Thread.sleep(2000);
 } catch (InterruptedException e) { e.printStackTrace();
 }
 System.out.println(線程 - +Thread.currentThread().getName()+ 結束執行...  + LocalDateTime.now());
}

總結

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

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

正文完