Redis分散式鎖解決微服務環境下定時任務問題【Aop+自定義註解實現】
阿新 • • 發佈：2021-01-23
技術標籤：Java開發經驗積累redisjava定時任務分散式鎖微服務
一、場景
定時任務，有過專案經歷的開發者估計都不陌生，是實現一些定時執行重複操作需求的常見解決方案。
在單機的情況下，定時任務當然是越用越爽，簡單粗暴直接cron表示式走起就行了，但是在微服務的場景下，要考慮多例項的問題。比如一個定時任務，由於被部署了在多臺機器上（或同一臺不同埠），這時候，可能會出現定時任務在同一時間被多次執行的問題。
為了保證在同一週期內，只有一個定時任務在執行，其他的不執行，可以採用redis分散式鎖、資料庫鎖、zookeeper鎖等方式去實現。
本文采用redis分散式鎖的思路去實現。
二、場景復現
建立一個springboot專案，匯入web依賴包，新建一個定時任務demo，每隔5秒執行一次，程式碼如下
/**
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/14 11:58
*/
@Component
@Slf4j
public class TaskDemo {
@Scheduled(cron = "0/5 * * * * *")
public void runTask(){
log.info("機器【1】上的 demo 定時任務啟動了！>> 當前時間 [{}]" 然後分別在 8090埠，和8091埠啟動這個專案。模擬多例項的情況，可以看到他們分別的執行結果如下：
A 伺服器
B 伺服器
如上可以看到，在同一個啟動時間點，兩臺伺服器上的同個定時任務都執行了。這顯然不太符合需求。
三、解決方案
使用redis分散式鎖
- 每個定時任務執行之前，先去redis那邊獲取鎖，如果獲取到了，則執行程式碼邏輯，獲取失敗則直接 return。
- 這個可以使用redis的setNX操作來實現，這個操作是原子性的，不過有個缺陷是沒有失效時間，這時如果伺服器A拿到鎖了，由於宕機或者其他網路不可達情況沒有釋放掉，則其他的伺服器永遠拿不到這個鎖，存在死鎖的情況。
- 所以這裡存redis的key是任務的名稱，value就是當前的時間戳+鎖過期時間。如果其他伺服器獲取鎖失敗了，看看上一個鎖是否已經過期。如果過期了則直接重新獲取鎖。
對於加鎖和解鎖操作，封裝成了工具類來實現，程式碼如下：
PS：這個獲取鎖的
getLock方法，第二個引數為鎖的過期時間，這裡一般設定為在任務預估執行時間和定時任務週期時間這個時間段內，如果設定過短，可能會導致鎖提前失效，任務還沒跑完的問題
/**
* 實現分散式redis鎖.
*
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/14 13:53
*/
@Component
public class RedisLockUtils {
/**
* 鎖名稱字首.
*/
public static final String TASK_LOCK_PREFIX = "TASK_LOCK_";
/**
* redis操作.
*/
@Autowired
private StringRedisTemplate redisTemplate;
/**
* 獲取分散式redis鎖.
* 邏輯：
* 1、使用setNX原子操作設定鎖（返回 true-代表加鎖成功，false-代表加鎖失敗）
* 2、加鎖成功直接返回
* 3、加鎖失敗，進行監測，是否存在死鎖的情況，檢查上一個鎖是否已經過期
* 4、如果過期，重新讓當前執行緒獲取新的鎖。
* 5、這裡可能會出現多個獲取鎖失敗的執行緒執行到這一步，所以判斷是否是加鎖成功，如果沒有，則返回失敗
*
* @param taskName 任務名稱
* @param lockExpireTime 鎖的過期時間
* @return true-獲取鎖成功 false-獲取鎖失敗
*/
public Boolean getLock(String taskName, long lockExpireTime) {
//鎖的名稱：字首 + 任務名稱
String lockName = TASK_LOCK_PREFIX + taskName;
return (Boolean) redisTemplate.execute((RedisCallback<?>) connection -> {
// 計算此次過期時間：當前時間往後延申一個expireTIme
long expireAt = System.currentTimeMillis() + lockExpireTime + 1;
// 獲取鎖（setNX 原子操作）
Boolean acquire = connection.setNX(lockName.getBytes(), String.valueOf(expireAt).getBytes());
// 如果設定成功
if (Objects.nonNull(acquire) && acquire) {
return true;
} else {
//防止死鎖，獲取舊的過期時間，與當前系統時間比是否過期，如果過期則允許其他的執行緒再次獲取。
byte[] bytes = connection.get(lockName.getBytes());
if (Objects.nonNull(bytes) && bytes.length > 0) {
long expireTime = Long.parseLong(new String(bytes));
// 如果舊的鎖已經過期
if (expireTime < System.currentTimeMillis()) {
// 重新讓當前執行緒加鎖
byte[] oldLockExpireTime = connection.getSet(lockName.getBytes(),
String.valueOf(System.currentTimeMillis() + lockExpireTime + 1).getBytes());
//這裡為null證明這個新鎖加鎖成功，上一個舊鎖已被釋放
if (Objects.isNull(oldLockExpireTime)) {
return true;
}
// 防止在併發場景下，被其他執行緒搶先加鎖成功的問題
return Long.parseLong(new String(oldLockExpireTime)) < System.currentTimeMillis();
}
}
}
return false;
});
}
/**
* 刪除鎖.
* 這裡可能會存在一種異常情況，即如果執行緒A加鎖成功
* 但是由於io或GC等原因在有效期內沒有執行完邏輯，這時執行緒B也可拿到鎖去執行。
* （方案，可以加鎖的時候新增當前執行緒的id作為標識，釋放鎖時，判斷一下，只能釋放自己加的鎖）
*
* @param lockName 鎖名稱
*/
public void delLock(String lockName) {
// 直接刪除key釋放redis鎖
redisTemplate.delete(lockName);
}
}
邏輯已經在註釋裡寫得很清楚了，這裡不再累述。
使用場景復現的demo來驗證
還是同樣的demo，不過這裡的定時任務邏輯可能要改動一下了，執行邏輯之前先去redis獲取一下鎖，只有獲取鎖成功了才可執行。
更改後的 demo 程式碼如下：
/**
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/14 11:58
*/
@Component
@Slf4j
public class TaskDemo {
@Autowired
private RedisLockUtils redisLockUtils;
@Scheduled(cron = "0/5 * * * * *")
public void start(){
try {
Boolean lock = redisLockUtils.getLock("test", 3000);
//獲取鎖
if (lock) {
log.info("機器【1】上的 demo 定時任務啟動了！>> 當前時間 [{}]", LocalDateTime.now());
//延遲一秒
Thread.sleep(1000);
}else {
log.error("獲取鎖失敗了，此次不執行！");
}
}catch (Exception e){
log.info("獲取鎖異常了！");
}finally {
//釋放redis鎖
redisLockUtils.delLock("test");
}
}
}
執行結果如下：
A 伺服器
B 伺服器
結果：同一個時間間隔裡只有一臺伺服器可以執行，與期望相符。
四、程式碼優化
現在的實現方式，必須要改動定時任務的程式碼邏輯，新增加鎖和解鎖的處理程式碼。這種與業務邏輯的耦合度有點高，不太美觀。這裡使用
自定義註解+AOP切面的方式來讓鎖處理 和 業務邏輯程式碼解耦，減少重複的程式碼。
關於自定義註解介紹，可以看我之前寫的一篇文章《Java使用自定義註解》
1、編寫自定義註解
編寫一個自定義註解，有兩個屬性，一個是定時任務名稱，用來標識這個鎖，一個是鎖的過期時間。
/**
* 自定義redis鎖註解.
* 目的：把加鎖解鎖邏輯與業務程式碼解耦.
*
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/14 16:50
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface TaskRedisLock {
/**
* 定時任務名稱.
*/
String taskName();
/**
* 定時任務鎖過期時間.
*/
long expireTime();
}
2、編寫AOP切面
編寫一個AOP切面，對上面定義的自定義註解進行攔截，然後獲取到裡面的屬性的值，最後再通過環繞通知來實現加鎖和解鎖的邏輯。
切面程式碼如下
/**
* 定時任務鎖切面，對加了自定義redis鎖註解的任務進行攔截.
*
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/14 16:59
*/
@Component
@Aspect
@Slf4j
public class RedisLockAspect {
//加鎖工具類
@Autowired
private RedisLockUtils redisLockUtils;
/**
* 攔截自定義的redis鎖註解.
*/
@Pointcut("@annotation(zpengblog.taskdemo.aop.custom.TaskRedisLock)")
public void pointCut(){}
/**
* 環繞通知.
*/
@Around("pointCut()")
public Object Around(ProceedingJoinPoint pjp) throws Throwable {
//獲取方法
MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
Method method = methodSignature.getMethod();
//獲取方法上的註解
TaskRedisLock annotation = method.getAnnotation(TaskRedisLock.class);
//獲取任務名稱
String taskName = annotation.taskName();
//獲取失效時間
long expireTime = annotation.expireTime();
try {
//獲取鎖
Boolean lock = redisLockUtils.getLock(taskName, expireTime);
if (lock) {
return pjp.proceed();
}else {
log.error("[{} 任務] 獲取redis鎖失敗了，此次不執行...", taskName);
}
}catch (Exception e){
log.error("[{} 任務]獲取鎖異常了！", taskName, e);
}finally {
//釋放redis鎖
redisLockUtils.delLock(taskName);
}
return null;
}
}
3、demo定時任務上使用該註解
/**
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/14 11:58
*/
@Component
@Slf4j
public class TaskDemo {
@Autowired
private RedisLockUtils redisLockUtils;
/**
* 使用自定義TaskRedisLock註解，通過aop來加鎖.
*/
@TaskRedisLock(taskName = "task_1", expireTime = 4000)
@Scheduled(cron = "0/5 * * * * *")
public void run(){
log.info("task_1 定時任務啟動了！>> 當前時間 [{}]", LocalDateTime.now());
try {
//延遲一秒
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
到此，成功把獲取redis鎖的邏輯和業務程式碼解耦，優化就完成啦。
PS：其實這個定時任務還是存在一些缺陷，目前我的場景基本滿足了，當然，設計一個高可用，完美的分散式鎖來說其實是一個複雜的事情。也有其他現成的方案可用，本文只是作為一個知識積累類共享，歡迎交流
https://blog.csdn.net/wang_jing_jing/article/details/106623112#comments_14608739 [參考資料]