SpringBoot運用AOP來實現(xiàn)分布式鎖的示例代碼

更新時間：2025年09月28日 09:14:33 作者：從零開始的-CodeNinja之路

本文主要介紹了通過注解和AOP實現(xiàn)分布式鎖的方案,包含鎖過期時間、等待超時設(shè)置及自動續(xù)約功能,利用定時任務(wù)監(jiān)控鎖狀態(tài)并延長有效期,感興趣的可以了解一下

一、鎖注解

我們首先寫一個鎖的注解

expiredTime 是設(shè)置鎖的過期時間，timeoutForLock 是設(shè)置等待鎖的超時時間。如果沒有等待獲得鎖的超時時間這個功能，那么其他線程在獲取鎖失敗時只能直接失敗，無法進行排隊等待。

/**
 * 分布式鎖注解
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
@Documented
public @interface RedisLock {
 
	long DEFAULT_TIMEOUT_FOR_LOCK = 5L;
	long DEFAULT_EXPIRE_TIME = 60L;
 
	String key() default "your-biz-key";
 
	long expiredTime() default DEFAULT_EXPIRE_TIME;
 
	long timeoutForLock() default DEFAULT_TIMEOUT_FOR_LOCK;
 
}

我們?nèi)绾问褂眠@個注解呢，很容易，在需要加鎖的業(yè)務(wù)方法上直接用就行.如下，我們有一個庫存服務(wù)類，它有一個扣減庫存方法，該方法將數(shù)據(jù)庫中的一個庫存商品的數(shù)量減一。在并發(fā)場景下，如果我們沒有對其進行資源控制，必然會發(fā)生庫存扣減不一致現(xiàn)象。

public class StockServiceImpl {
	@RedisLock(key = "stock-lock", expiredTime = 10L, timeoutForLock = 5L)
	public void deduct(Long stockId) {
		Stock stock = this.getById(1L);
		Integer count = stock.getCount();
		stock.setCount(count - 1);
		this.updateById(stock);
	}
}

二、在 AOP 中加鎖處理

我們需要使用 AOP 來處理什么？自然是處理使用@RedisLock的方法，因此我們寫一個切點表達式，它匹配所有標有 @RedisLock 注解的方法。

接著，我們將此切點表達式與 @Around 注解結(jié)合使用，以創(chuàng)建環(huán)繞通知，在目標方法執(zhí)行前后執(zhí)行我們的加鎖解鎖邏輯。

因此，基本的邏輯我們就理清了，代碼大致長下面這個樣子:

public class RedisLockAspect {
 
	private final RedisTemplate<String, Object> redisTemplate;
 
	// 鎖的redis key前綴
	private static final String DEFAULT_KEY_PREFIX = "lock:";
 
	// 匹配所有標有 @RedisLock 注解的方法
	@Pointcut("@annotation(com.kelton.lock.annotation.RedisLock)")
	public void lockAnno() {
	}
 
 
	@Around("lockAnno()")
	public void invoke(ProceedingJoinPoint joinPoint) throws Exception {
		// 獲取攔截方法上的RedisLock注解
		RedisLock annotation = getLockAnnotationOnMethod(joinPoint);
		// 獲取鎖key
		String key = getKey(annotation);
		// 鎖過期時間
		long expireTime = annotation.expiredTime();
		// 獲取鎖的等待時間
		long timeoutForLock = annotation.timeoutForLock();
		// 在這里加鎖
		someCodeForLock...
		// 執(zhí)行業(yè)務(wù)
		joinPoint.proceed();
		// 在這里解鎖
		someCodeForUnLock...
	}

我們在加鎖的時候，需要用上 timeoutForLock 這個屬性，我們通過自旋加線程休眠的方式，來達到在一段時間內(nèi)等待獲取鎖的目的。如果自旋時間結(jié)束后，還沒獲取鎖，則拋出異常，這里可以根據(jù)自己情況而定。自旋加鎖代碼如下：

// 自旋獲取鎖
long endTime = System.currentTimeMillis() + timeoutForLock * 1000;
boolean acquired = false;
String uuid = UUID.randomUUID().toString();
while(System.currentTimeMillis() < endTime) {
    Boolean absent = redisTemplate.opsForValue()
            .setIfAbsent(key, uuid, expireTime, TimeUnit.SECONDS);
 
    if (Boolean.TRUE.equals(absent)) {
        acquired = true;
        break;
    } else {
        // 獲取不到鎖，嘗試休眠100毫秒后重試
        Thread.sleep(100);
    }
}
// 超時未獲取到鎖， 拋出異常，可根據(jù)自己業(yè)務(wù)而定
if (!acquired) {
    throw new RuntimeException("獲取鎖異常");
}

我們發(fā)現(xiàn)上面加鎖的時候設(shè)置了一個 uuid 作為 value 值，這是為了在鎖釋放的時候，不誤刪其他線程上的鎖，隨后，我們就可以執(zhí)行被 AOP 切中的方法，執(zhí)行結(jié)束釋放鎖。代碼如下：

try {
    // 執(zhí)行業(yè)務(wù)
    joinPoint.proceed();
} catch (Throwable e) {
    log.error("業(yè)務(wù)執(zhí)行出錯！");
} finally {
    // 解鎖時進行校驗，只刪除自己線程加的鎖
    String value = (String) redisTemplate.opsForValue().get(key);
    if (uuid.equals(value)) {
        redisTemplate.delete(key);
    } else {
        log.warn("鎖已過期！");
    }
}

到這里，我們就以注解+AOP 的方式實現(xiàn)了分布式鎖的功能。當(dāng)然，以上只實現(xiàn)了分布式鎖的簡單功能，還缺少了分布式鎖的 key 自動續(xù)約防止鎖過期功能，以及鎖重入功能。

目前，RedisLockAspect的完整代碼如下：

@Component
@Aspect
@Slf4j
@AllArgsConstructor
public class RedisLockAspect {
 
	// 匹配所有標有 @RedisLock 注解的方法
	@Pointcut("@annotation(com.kelton.lock.annotation.RedisLock)")
	public void lockAnno() {
	}
 
 
	@Around("lockAnno()")
	public void invoke(ProceedingJoinPoint joinPoint) throws Exception {
		// 獲取攔截方法上的RedisLock注解
		RedisLock annotation = getLockAnnotationOnMethod(joinPoint);
 
		String key = getKey(annotation);
		// 鎖過期時間
		long expireTime = annotation.expiredTime();
		// 獲取鎖的等待時間
		long timeoutForLock = annotation.timeoutForLock();
		// 自旋獲取鎖
		long endTime = System.currentTimeMillis() + timeoutForLock * 1000;
		boolean acquired = false;
		String uuid = UUID.randomUUID().toString();
		while(System.currentTimeMillis() < endTime) {
			Boolean absent = redisTemplate.opsForValue()
					.setIfAbsent(key, uuid, expireTime, TimeUnit.SECONDS);
 
			if (Boolean.TRUE.equals(absent)) {
				acquired = true;
				break;
			} else {
				// 獲取不到鎖，嘗試休眠100毫秒后重試
				Thread.sleep(100);
			}
		}
		// 超時未獲取到鎖， 拋出異常，可根據(jù)自己業(yè)務(wù)而定
		if (!acquired) {
			throw new RuntimeException("獲取鎖異常");
		}
		try {
			// 執(zhí)行業(yè)務(wù)
			joinPoint.proceed();
		} catch (Throwable e) {
			log.error("業(yè)務(wù)執(zhí)行出錯！");
		} finally {
			// 解鎖時進行校驗，只刪除自己線程加的鎖
			String value = (String) redisTemplate.opsForValue().get(key);
			if (uuid.equals(value)) {
				redisTemplate.delete(key);
			} else {
				log.warn("鎖已過期！");
			}
		}
	}
 
	private String getKey(RedisLock redisLock) {
		if (Objects.isNull(redisLock)) {
			return DEFAULT_KEY_PREFIX + "default";
		}
		return DEFAULT_KEY_PREFIX + redisLock.key();
	}
 
	private RedisLock getLockAnnotationOnMethod(ProceedingJoinPoint joinPoint) {
		MethodSignature signature = (MethodSignature) joinPoint.getSignature();
		Method method = signature.getMethod();
		return method.getAnnotation(RedisLock.class);
	}
 
}

三、 key 自動續(xù)約防止鎖過期

我們接著完善該分布式鎖，為其添加 key 自動續(xù)約防止鎖過期的功能。我們的思路與Redission的watch dog類似，開啟一個后臺線程，來定時檢查需要續(xù)約的鎖。

我們?nèi)绾闻袛嘁粋€鎖是否需要續(xù)約呢，我們可以簡單定義一個續(xù)約分界線，比如在鎖過期時間的三分之二的時間點及之后，對鎖進行續(xù)約。

3.1 定義一個續(xù)約任務(wù)

我們來定義一個鎖續(xù)約任務(wù)，那我們需要什么信息呢？

我們至少需要鎖的 key，鎖要設(shè)置的過期時間。這是兩個最基本的信息。

要判斷在鎖過期時間的三分之二的時間點及之后進行續(xù)約，那么我們還需要記錄鎖上次續(xù)約的時間點。

此外，我們還可以為鎖續(xù)約任務(wù)添加最大續(xù)約次數(shù)限制，這可以避免某些執(zhí)行時間特別久的任務(wù)不斷占用鎖。

所以我們還需要記錄當(dāng)前鎖續(xù)約次數(shù)和最大續(xù)約次數(shù)。

對超過最大續(xù)約次數(shù)的鎖的線程，我們直接將其停止，因此我們也記錄一下該鎖的線程。

結(jié)合上面的分析，我們定義的鎖續(xù)約任務(wù)類如下：

public class LockRenewTask {
 
	/**
	 * key
	 */
	private final String key;
	/**
	 * 過期時間。單位：秒
	 */
	private final long expiredTime;
	/**
	 * 鎖的最大續(xù)約次數(shù)
	 */
	private final int maxRenewCount;
	/**
	 * 鎖的當(dāng)前續(xù)約次數(shù)
	 */
	private int currentRenewCount;
	/**
	 * 最新更新時間
	 */
	private LocalDateTime latestRenewTime;
	/**
	 * 業(yè)務(wù)線程
	 */
	private final Thread thread;
 
	public LockRenewTask(String key, long expiredTime, int maxRenewCount, Thread thread) {
		this.key = key;
		this.expiredTime = expiredTime;
		this.maxRenewCount = maxRenewCount;
		this.thread = thread;
		this.latestRenewTime = LocalDateTime.now();
	}
	/**
	 * 是否到達續(xù)約時間
	 * @return
	 */
	public boolean isTimeToRenew() {
		LocalDateTime now = LocalDateTime.now();
		Duration duration = Duration.between(latestRenewTime, now);
 
		return duration.toSeconds() >= ((double)(this.expiredTime / 3) * 2);
	}
	/**
	 * 是否達到最大續(xù)約次數(shù)
	 * @return
	 */
	public boolean exceedMaxRenewCount() {
		return this.currentRenewCount >= this.maxRenewCount;
	}
	public synchronized void renew() {
		this.currentRenewCount++;
		this.latestRenewTime = LocalDateTime.now();
	}
	// 取消業(yè)務(wù)方法
	public void cancel() {
		thread.interrupt();
	}
	public String getKey() {
		return key;
	}
	public long getExpiredTime() {
		return expiredTime;
	}
}

isTimeToRenew(): 判斷是否可以對鎖進行續(xù)約
exceedMaxRenewCount(): 判斷是否達到最大續(xù)約次數(shù)
renew(): 來標記一次續(xù)約操作
cancel(): 取消業(yè)務(wù)方法

3.2 定義一個鎖續(xù)約任務(wù)處理器

接著，我們定義一個定時執(zhí)行該續(xù)約任務(wù)的 handler。

該 handler 也比較簡答，核心邏輯是持有一個類型為 List的 taskList 來添加續(xù)約任務(wù)，且使用一個 ScheduledExecutorService 來定時遍歷該 taskList 來執(zhí)行續(xù)約任務(wù)。該 handler 再對外暴露一個 addRenewTask 方法，方便外部調(diào)用來添加續(xù)約任務(wù)到 taskList 中。

@Slf4j
@Component
public class LockRenewHandler {
 
	@Autowired
	private RedisTemplate<String, Object> redisTemplate;
 
	/**
	 * 保障對 taskList的添加刪除操作是線程安全的
	 */
	private final ReentrantLock taskListLock = new ReentrantLock();
 
	private final List<LockRenewTask> taskList = new ArrayList<>();
 
	private final ScheduledExecutorService taskExecutorService;
 
	{
		taskExecutorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
		taskExecutorService.scheduleAtFixedRate(() -> {
			try {
				executeRenewTask();
			} catch (Exception e) {
				//錯誤處理
			}
		}, 1, 2, TimeUnit.SECONDS);
 
	}
	/**
	 * 添加續(xù)約任務(wù)
	 */
	public void addRenewTask(LockRenewTask task) {
		taskListLock.lock();
		try {
			taskList.add(task);
		} finally {
			taskListLock.unlock();
		}
	}
	/**
	 * 執(zhí)行續(xù)約任務(wù)
	 */
	private void executeRenewTask() {
		log.info("開始執(zhí)行續(xù)約任務(wù)");
		if (CollectionUtils.isEmpty(taskList)) {
			return;
		}
		// 需要刪除的任務(wù)，暫存這個集合中  取消
		List<LockRenewTask> cancelTask = new ArrayList<>();
		// 獲取任務(wù)副本
		List<LockRenewTask> copyTaskList = new ArrayList<>(taskList);
		for (LockRenewTask task : copyTaskList) {
			try {
				// 判斷 Redis 中是否存在 key
				if (!redisTemplate.hasKey(task.getKey())) {
					cancelTask.add(task);
					continue;
				}
				// 大于等于最大續(xù)約次數(shù)
				if (task.exceedMaxRenewCount()) {
					// 停止續(xù)約任務(wù)
					task.cancel();
					cancelTask.add(task);
					continue;
				}
				// 到達續(xù)約時間
				if (task.isTimeToRenew()) {
					log.info("續(xù)約任務(wù)：{}", task.getKey());
					redisTemplate.expire(task.getKey(), task.getExpiredTime(), TimeUnit.SECONDS);
					task.renew();
				}
			} catch (Exception e) {
				//錯誤處理
				log.error("處理任務(wù)出錯：{}", task);
			}
		}
		// 加鎖，刪除 taskList 中需要移除的任務(wù)
		taskListLock.lock();
		try {
			taskList.removeAll(cancelTask);
			// 清理cancelTask,避免堆積，產(chǎn)生內(nèi)存泄露
			cancelTask.clear();
		} finally {
			taskListLock.unlock();
		}
	}
}

總結(jié)一下 LockRenewHandler的主要作用：它負責(zé)管理和執(zhí)行續(xù)約任務(wù)，以延長 Redis 中鍵的過期時間。

添加續(xù)約任務(wù)：addRenewTask() 方法允許添加新的續(xù)約任務(wù)到內(nèi)部列表 taskList 中。
執(zhí)行續(xù)約任務(wù)：executeRenewTask() 方法定期執(zhí)行續(xù)約任務(wù)。它檢查每個任務(wù)的狀態(tài)，并根據(jù)需要續(xù)約 Redis 中的鍵。
移除完成的任務(wù)：維護一個 cancelTask 列表，用于存儲需要從 taskList 中移除的任務(wù)。
在 executeRenewTask() 方法中，它會將完成的任務(wù)添加到 cancelTask 列表中，并在之后將其從 taskList
中移除。

大概的工作流程如下：

續(xù)約任務(wù)被添加到 taskList 中。
executeRenewTask() 方法定期執(zhí)行，它檢查每個任務(wù)的狀態(tài)：
- 如果 Redis 中不再存在該鍵，則取消任務(wù)。
- 如果任務(wù)的續(xù)約次數(shù)達到上限，則取消任務(wù)。
- 如果是時候續(xù)約了，則續(xù)約 Redis 中的鍵并更新任務(wù)的續(xù)約次數(shù)，記錄續(xù)約時間點。
完成的任務(wù)被添加到 cancelTask 列表中。
executeRenewTask() 方法獲取 taskList 的副本，并從副本中移除 cancelTask
中的任務(wù)，并且在完成移除任務(wù)操作后清空cancelTask。
更新后的 taskList 被保存回類中。

兩個需要注意的點

我們遍歷taskList時拷貝了一份副本進行遍歷，因為taskList是可變的，這樣可以避免在遍歷的時候產(chǎn)生并發(fā)修改問題。

cancelTask需要清理，避免產(chǎn)生內(nèi)存泄漏。

通過這種方式，LockRenewHandler 可以確保 Redis 中的鍵在需要時得到續(xù)約，并自動移除完成或失敗的任務(wù)。

3.3 添加鎖續(xù)約任務(wù)

在上面 3.1 節(jié)和 3.2 節(jié)我們定義好了鎖續(xù)約任務(wù)和處理鎖續(xù)約任務(wù)的核心代碼，接下來我們需要在第 2 節(jié)加鎖解鎖的 AOP 處理邏輯上進行一點小小的修改，主要就是在執(zhí)行加鎖之后，執(zhí)行業(yè)務(wù)代碼之前，添加上鎖續(xù)約任務(wù)。修改位置如下：

public void invoke(ProceedingJoinPoint joinPoint) throws Exception {
    ... // 省略代碼
    try {
        // 添加鎖續(xù)約任務(wù)
        LockRenewTask task = new LockRenewTask(key, annotation.expiredTime(), annotation.maxRenew(), Thread.currentThread());
        lockRenewHandler.addRenewTask(task);
        log.info("添加續(xù)約任務(wù), key:{}", key);
        // 執(zhí)行業(yè)務(wù)
        joinPoint.proceed();
    } catch (Throwable e) {
        log.error("業(yè)務(wù)執(zhí)行出錯！");
    } finally {
        // 解鎖時進行校驗，只刪除自己線程加的鎖
        String value = (String) redisTemplate.opsForValue().get(key);
        if (uuid.equals(value)) {
            redisTemplate.delete(key);
        } else {
            log.warn("鎖已過期！");
        }
    }
    ... // 省略代碼
}

到這里，我們的分布式鎖已經(jīng)相當(dāng)完善了，把鎖自動續(xù)約的功能也加上了。當(dāng)然，還沒有實現(xiàn)鎖的可重入性。

到此這篇關(guān)于SpringBoot運用AOP來實現(xiàn)分布式鎖的文章就介紹到這了,更多相關(guān)SpringBoot AOP分布式鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: