使用注解優(yōu)雅實(shí)現(xiàn)分布式鎖，你學(xué)會了嗎？
1. 業(yè)務(wù)背景
有些業(yè)務(wù)請求，屬于耗時操作，需要加鎖，防止后續(xù)的并發(fā)操作，同時對數(shù)據(jù)庫的數(shù)據(jù)進(jìn)行操作，需要避免對之前的業(yè)務(wù)造成影響。
2. 分析流程
使用 Redis 作為分布式鎖，將鎖的狀態(tài)放到 Redis 統(tǒng)一維護(hù)，解決集群中單機(jī) JVM 信息不互通的問題，規(guī)定操作順序，保護(hù)用戶的數(shù)據(jù)正確。
梳理設(shè)計流程
- 新建注解 @interface，在注解里設(shè)定入?yún)?biāo)志
- 增加 AOP 切點(diǎn)，掃描特定注解
- 建立 @Aspect 切面任務(wù)，注冊 bean 和攔截特定方法
- 特定方法參數(shù) ProceedingJoinPoint，對方法 pjp.proceed() 前后進(jìn)行攔截
- 切點(diǎn)前進(jìn)行加鎖，任務(wù)執(zhí)行后進(jìn)行刪除 key
核心步驟：加鎖、解鎖和續(xù)時
加鎖
使用了 RedisTemplate 的 opsForValue.setIfAbsent 方法，判斷是否有 key，設(shè)定一個隨機(jī)數(shù) UUID.random().toString，生成一個隨機(jī)數(shù)作為 value。
從 redis 中獲取鎖之后，對 key 設(shè)定 expire 失效時間，到期后自動釋放鎖。
按照這種設(shè)計，只有第一個成功設(shè)定 Key 的請求，才能進(jìn)行后續(xù)的數(shù)據(jù)操作，后續(xù)其它請求由于無法獲得鎖資源，將會失敗結(jié)束。
超時問題
擔(dān)心 pjp.proceed() 切點(diǎn)執(zhí)行的方法太耗時，導(dǎo)致 Redis 中的 key 由于超時提前釋放了。
例如，線程 A 先獲取鎖，proceed 方法耗時，超過了鎖超時時間，到期釋放了鎖，這時另一個線程 B 成功獲取 Redis 鎖，兩個線程同時對同一批數(shù)據(jù)進(jìn)行操作，導(dǎo)致數(shù)據(jù)不準(zhǔn)確。
解決方案：增加一個「續(xù)時」
任務(wù)不完成，鎖不釋放：
維護(hù)了一個定時線程池 ScheduledExecutorService，每隔 2s 去掃描加入隊列中的 Task，判斷是否失效時間是否快到了，公式為：【失效時間】<= 【當(dāng)前時間】+【失效間隔（三分之一超時）】
/**
* 線程池，每個 JVM 使用一個線程去維護(hù) keyAliveTime，定時執(zhí)行 runnable
*/
private static final ScheduledExecutorService SCHEDULER =
new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().namingPattern("redisLock-schedule-pool").daemon(true).build());
static {
SCHEDULER.scheduleAtFixedRate(() -> {
// do something to extend time
}, 0, 2, TimeUnit.SECONDS);
}3. 設(shè)計方案
經(jīng)過上面的分析，設(shè)計出了這個方案：
圖片
前面已經(jīng)說了整體流程，這里強(qiáng)調(diào)一下幾個核心步驟：
- 攔截注解 @RedisLock，獲取必要的參數(shù)
- 加鎖操作
- 續(xù)時操作
- 結(jié)束業(yè)務(wù)，釋放鎖
4. 實(shí)操
之前也有整理過 AOP 使用方法，可以參考一下
相關(guān)屬性類配置
業(yè)務(wù)屬性枚舉設(shè)定
public enum RedisLockTypeEnum {
/**
* 自定義 key 前綴
*/
ONE("Business1", "Test1"),
TWO("Business2", "Test2");
private String code;
private String desc;
RedisLockTypeEnum(String code, String desc) {
this.code = code;
this.desc = desc;
}
public String getCode() {
return code;
}
public String getDesc() {
return desc;
}
public String getUniqueKey(String key) {
return String.format("%s:%s", this.getCode(), key);
}
}任務(wù)隊列保存參數(shù)
public class RedisLockDefinitionHolder {
/**
* 業(yè)務(wù)唯一 key
*/
private String businessKey;
/**
* 加鎖時間 (秒 s)
*/
private Long lockTime;
/**
* 上次更新時間（ms）
*/
private Long lastModifyTime;
/**
* 保存當(dāng)前線程
*/
private Thread currentTread;
/**
* 總共嘗試次數(shù)
*/
privateint tryCount;
/**
* 當(dāng)前嘗試次數(shù)
*/
privateint currentCount;
/**
* 更新的時間周期（毫秒）,公式 = 加鎖時間（轉(zhuǎn)成毫秒） / 3
*/
private Long modifyPeriod;
public RedisLockDefinitionHolder(String businessKey, Long lockTime, Long lastModifyTime, Thread currentTread, int tryCount) {
this.businessKey = businessKey;
this.lockTime = lockTime;
this.lastModifyTime = lastModifyTime;
this.currentTread = currentTread;
this.tryCount = tryCount;
this.modifyPeriod = lockTime * 1000 / 3;
}
}設(shè)定被攔截的注解名字
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.TYPE})
public @interface RedisLockAnnotation {
/**
* 特定參數(shù)識別，默認(rèn)取第 0 個下標(biāo)
*/
int lockFiled() default 0;
/**
* 超時重試次數(shù)
*/
int tryCount() default 3;
/**
* 自定義加鎖類型
*/
RedisLockTypeEnum typeEnum();
/**
* 釋放時間，秒 s 單位
*/
long lockTime() default 30;
}核心切面攔截的操作
RedisLockAspect.java 該類分成三部分來描述具體作用
Pointcut 設(shè)定
/**
* @annotation 中的路徑表示攔截特定注解
*/
@Pointcut("@annotation(cn.sevenyuan.demo.aop.lock.RedisLockAnnotation)")
public void redisLockPC() {
}Around 前后進(jìn)行加鎖和釋放鎖
前面步驟定義了我們想要攔截的切點(diǎn)，下一步就是在切點(diǎn)前后做一些自定義操作：
@Around(value = "redisLockPC()")
public Object around(ProceedingJoinPoint pjp) throws Throwable {
// 解析參數(shù)
Method method = resolveMethod(pjp);
RedisLockAnnotation annotation = method.getAnnotation(RedisLockAnnotation.class);
RedisLockTypeEnum typeEnum = annotation.typeEnum();
Object[] params = pjp.getArgs();
String ukString = params[annotation.lockFiled()].toString();
// 省略很多參數(shù)校驗和判空
String businessKey = typeEnum.getUniqueKey(ukString);
String uniqueValue = UUID.randomUUID().toString();
// 加鎖
Object result = null;
try {
boolean isSuccess = redisTemplate.opsForValue().setIfAbsent(businessKey, uniqueValue);
if (!isSuccess) {
thrownew Exception("You can't do it，because another has get the lock =-=");
}
redisTemplate.expire(businessKey, annotation.lockTime(), TimeUnit.SECONDS);
Thread currentThread = Thread.currentThread();
// 將本次 Task 信息加入「延時」隊列中
holderList.add(new RedisLockDefinitionHolder(businessKey, annotation.lockTime(), System.currentTimeMillis(),
currentThread, annotation.tryCount()));
// 執(zhí)行業(yè)務(wù)操作
result = pjp.proceed();
// 線程被中斷，拋出異常，中斷此次請求
if (currentThread.isInterrupted()) {
thrownew InterruptedException("You had been interrupted =-=");
}
} catch (InterruptedException e ) {
log.error("Interrupt exception, rollback transaction", e);
thrownew Exception("Interrupt exception, please send request again");
} catch (Exception e) {
log.error("has some error, please check again", e);
} finally {
// 請求結(jié)束后，強(qiáng)制刪掉 key，釋放鎖
redisTemplate.delete(businessKey);
log.info("release the lock, businessKey is [" + businessKey + "]");
}
return result;
}上述流程簡單總結(jié)一下：
- 解析注解參數(shù)，獲取注解值和方法上的參數(shù)值
- redis 加鎖并且設(shè)置超時時間
- 將本次 Task 信息加入「延時」隊列中，進(jìn)行續(xù)時，方式提前釋放鎖
- 加了一個線程中斷標(biāo)志
- 結(jié)束請求，finally 中釋放鎖
續(xù)時操作
這里用了 ScheduledExecutorService，維護(hù)了一個線程，不斷對任務(wù)隊列中的任務(wù)進(jìn)行判斷和延長超時時間：
// 掃描的任務(wù)隊列
privatestatic ConcurrentLinkedQueue<RedisLockDefinitionHolder> holderList = new ConcurrentLinkedQueue();
/**
* 線程池，維護(hù)keyAliveTime
*/
privatestaticfinal ScheduledExecutorService SCHEDULER = new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().namingPattern("redisLock-schedule-pool").daemon(true).build());
{
// 兩秒執(zhí)行一次「續(xù)時」操作
SCHEDULER.scheduleAtFixedRate(() -> {
// 這里記得加 try-catch，否者報錯后定時任務(wù)將不會再執(zhí)行=-=
Iterator<RedisLockDefinitionHolder> iterator = holderList.iterator();
while (iterator.hasNext()) {
RedisLockDefinitionHolder holder = iterator.next();
// 判空
if (holder == null) {
iterator.remove();
continue;
}
// 判斷 key 是否還有效，無效的話進(jìn)行移除
if (redisTemplate.opsForValue().get(holder.getBusinessKey()) == null) {
iterator.remove();
continue;
}
// 超時重試次數(shù)，超過時給線程設(shè)定中斷
if (holder.getCurrentCount() > holder.getTryCount()) {
holder.getCurrentTread().interrupt();
iterator.remove();
continue;
}
// 判斷是否進(jìn)入最后三分之一時間
long curTime = System.currentTimeMillis();
boolean shouldExtend = (holder.getLastModifyTime() + holder.getModifyPeriod()) <= curTime;
if (shouldExtend) {
holder.setLastModifyTime(curTime);
redisTemplate.expire(holder.getBusinessKey(), holder.getLockTime(), TimeUnit.SECONDS);
log.info("businessKey : [" + holder.getBusinessKey() + "], try count : " + holder.getCurrentCount());
holder.setCurrentCount(holder.getCurrentCount() + 1);
}
}
}, 0, 2, TimeUnit.SECONDS);
}這段代碼，用來實(shí)現(xiàn)設(shè)計圖中虛線框的思想，避免一個請求十分耗時，導(dǎo)致提前釋放了鎖。
這里加了「線程中斷」Thread#interrupt，希望超過重試次數(shù)后，能讓線程中斷（未經(jīng)嚴(yán)謹(jǐn)測試，僅供參考哈哈哈哈）
不過建議如果遇到這么耗時的請求，還是能夠從根源上查找，分析耗時路徑，進(jìn)行業(yè)務(wù)優(yōu)化或其它處理，避免這些耗時操作。
所以記得多打點(diǎn) Log，分析問題時可以更快一點(diǎn)。記錄項目日志，一個注解搞定
5. 開始測試
在一個入口方法中，使用該注解，然后在業(yè)務(wù)中模擬耗時請求，使用了 Thread#sleep
@GetMapping("/testRedisLock")
@RedisLockAnnotation(typeEnum = RedisLockTypeEnum.ONE, lockTime = 3)
public Book testRedisLock(@RequestParam("userId") Long userId) {
try {
log.info("睡眠執(zhí)行前");
Thread.sleep(10000);
log.info("睡眠執(zhí)行后");
} catch (Exception e) {
// log error
log.info("has some error", e);
}
return null;
}使用時，在方法上添加該注解，然后設(shè)定相應(yīng)參數(shù)即可，根據(jù) typeEnum 可以區(qū)分多種業(yè)務(wù)，限制該業(yè)務(wù)被同時操作。
測試結(jié)果：
2020-04-04 14:55:50.864 INFO 9326 --- [nio-8081-exec-1] c.s.demo.controller.BookController : 睡眠執(zhí)行前
2020-04-04 14:55:52.855 INFO 9326 --- [k-schedule-pool] c.s.demo.aop.lock.RedisLockAspect : businessKey : [Business1:1024], try count : 0
2020-04-04 14:55:54.851 INFO 9326 --- [k-schedule-pool] c.s.demo.aop.lock.RedisLockAspect : businessKey : [Business1:1024], try count : 1
2020-04-04 14:55:56.851 INFO 9326 --- [k-schedule-pool] c.s.demo.aop.lock.RedisLockAspect : businessKey : [Business1:1024], try count : 2
2020-04-04 14:55:58.852 INFO 9326 --- [k-schedule-pool] c.s.demo.aop.lock.RedisLockAspect : businessKey : [Business1:1024], try count : 3
2020-04-04 14:56:00.857 INFO 9326 --- [nio-8081-exec-1] c.s.demo.controller.BookController : has some error
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method) [na:1.8.0_221]我這里測試的是重試次數(shù)過多，失敗的場景，如果減少睡眠時間，就能讓業(yè)務(wù)正常執(zhí)行。
如果同時請求，你將會發(fā)現(xiàn)以下錯誤信息：
圖片
表示我們的鎖??的確生效了，避免了重復(fù)請求。
6. 總結(jié)
對于耗時業(yè)務(wù)和核心數(shù)據(jù)，不能讓重復(fù)的請求同時操作數(shù)據(jù)，避免數(shù)據(jù)的不正確，所以要使用分布式鎖來對它們進(jìn)行保護(hù)。
再來梳理一下設(shè)計流程：
- 新建注解 @interface，在注解里設(shè)定入?yún)?biāo)志
- 增加 AOP 切點(diǎn)，掃描特定注解
- 建立 @Aspect 切面任務(wù)，注冊 bean 和攔截特定方法
- 特定方法參數(shù) ProceedingJoinPoint，對方法 pjp.proceed() 前后進(jìn)行攔截
- 切點(diǎn)前進(jìn)行加鎖，任務(wù)執(zhí)行后進(jìn)行刪除 key
本次學(xué)習(xí)是通過 Review 小伙伴的代碼設(shè)計，從中了解分布式鎖的具體實(shí)現(xiàn)，仿照他的設(shè)計，重新寫了一份簡化版的業(yè)務(wù)處理。對于之前沒考慮到的「續(xù)時」操作，這里使用了守護(hù)線程來定時判斷和延長超時時間，避免了鎖提前釋放。
