作者：
蔡柱梁
現(xiàn)在的服務(wù)往往都是多節(jié)點(diǎn)，在一些特定的場景下容易產(chǎn)生并發(fā)問題，比如扣減庫存，送完即止活動(dòng)，中臺(tái)的批量導(dǎo)入（有唯一校驗(yàn)要求）等等。這時(shí)，我們可以通過分布式鎖解決這些問題。
作者 | 蔡柱梁
審校 | 重樓
目錄
- 分布式鎖介紹
- 如何實(shí)現(xiàn)分布式鎖
- 實(shí)現(xiàn)分布式鎖
1 分布式鎖介紹
現(xiàn)在的服務(wù)往往都是多節(jié)點(diǎn)，在一些特定的場景下容易產(chǎn)生并發(fā)問題，比如扣減庫存，送完即止活動(dòng)，中臺(tái)的批量導(dǎo)入（有唯一校驗(yàn)要求）等等。這時(shí)，我們可以通過分布式鎖解決這些問題。
2 如何實(shí)現(xiàn)分布式鎖
實(shí)現(xiàn)的方式有很多種，如：
- 基于 MySQL 等數(shù)據(jù)庫實(shí)現(xiàn)
- 基于 ZooKeeper 實(shí)現(xiàn)
- 基于 Redis 實(shí)現(xiàn)不管采用什么技術(shù)棧實(shí)現(xiàn)，但是邏輯流程都是大體不差的。下面是筆者自己在工作中基于Redis 實(shí)踐過的流程圖：
3 實(shí)現(xiàn)分布式鎖
其實(shí)可以不用自己手寫，現(xiàn)在有一個(gè)中間件Redisson 相當(dāng)好用，十分推薦。這里的實(shí)現(xiàn)更多是用于學(xué)習(xí)。
3.1 Redis 是單節(jié)點(diǎn)的情況下實(shí)現(xiàn)的分布式鎖
需要使用分布式鎖的業(yè)務(wù)代碼如下：
package com.example.demo.test.utils;
import com.example.demo.utils.RedisLockUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
/**
* @author CaiZhuliang
* @date 2023/8/31
*/
@Slf4j
@SpringBootTest
public class RedisLockUtilTest {
@Autowired
private RedisLockUtil redisLockUtil;
@Test
public void simpleLockTest() {
String key = "redis:lock:" + System.currentTimeMillis();
boolean result = redisLockUtil.lock(key, 8_000L);
if (result) {
try {
// do something
} catch (Exception e) {
log.error("simpleLockTest - 系統(tǒng)異常!", e);
} finally {
boolean unlock = redisLockUtil.unlock(key);
if (!unlock) {
log.error("simpleLockTest - 釋放鎖失敗，key : {}", key);
}
}
}
}
}分布式鎖工具類代碼如下：
package com.example.demo.utils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author CaiZhuliang
* @date 2023/8/31
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisLockUtil {
private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
new BasicThreadFactory.Builder()
.namingPattern("redisLockUtil-schedule-pool-%d")
.daemon(true)
.build());
private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<>();
private final RedisTemplate<String, String> redisTemplate;
/**
* 釋放鎖
* <p>必須和RedisLockUtil#simpleLock是同一個(gè)線程</p>
* @param key 需要釋放鎖的key
* @return true-成功 false-失敗
*/
public boolean releaseSimpleLock(String key) {
String token = THREAD_LOCAL.get();
try {
String remoteToken = redisTemplate.opsForValue().get(key);
if (!token.equals(remoteToken)) {
// 當(dāng)前線程不再持有鎖
return false;
}
// 是自己持有鎖才能釋放
return Boolean.TRUE.equals(redisTemplate.delete(key));
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}", key, e);
return false;
} finally {
THREAD_LOCAL.remove();
}
}
/**
* 這個(gè)方法不考慮Redis的集群架構(gòu)，不考慮腦裂問題，當(dāng)只有一個(gè)Redis來考慮。
* @param key 需要上鎖的key
* @param expireTime 過期時(shí)間，單位：毫秒
* @return true-成功 false-失敗
*/
public boolean simpleLock(String key, Long expireTime) {
if (StringUtils.isBlank(key)) {
log.warn("非cluster模式簡單分布式鎖 - key is blank");
return false;
}
if (null == expireTime || expireTime <= 0) {
expireTime = 0L;
}
String token = UUID.randomUUID().toString();
// 續(xù)約周期，單位納秒
long renewPeriod = expireTime / 2 * 1000_000;
try {
// 設(shè)置鎖
Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
if (Boolean.FALSE.equals(result)) {
return false;
}
// 上鎖成功后將令牌綁定當(dāng)前線程
THREAD_LOCAL.set(token);
if (renewPeriod > 0) {
// 續(xù)約任務(wù)
renewTask(key, token, expireTime, renewPeriod);
}
return true;
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 上鎖失敗。", e);
THREAD_LOCAL.remove();
return false;
}
}
/**
* 鎖續(xù)約任務(wù)
* @param key 需要續(xù)命的key
* @param token 成功獲鎖的線程持有的令牌
* @param expireTime 過期時(shí)間，單位：毫秒
* @param renewPeriod 續(xù)約周期，單位：納秒
*/
private void renewTask(String key, String token, long expireTime, long renewPeriod) {
EXECUTOR_SERVICE.schedule(() -> {
ValueOperations<String, String> valueOperator = redisTemplate.opsForValue();
String val = valueOperator.get(key);
if (token.equals(val)) {
// 是自己持有鎖才能續(xù)約
try {
Boolean result = valueOperator.setIfPresent(key, token, expireTime, TimeUnit.MILLISECONDS);
if (Boolean.TRUE.equals(result)) {
// 續(xù)約成功
log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}", key);
// 開啟下一次續(xù)約任務(wù)
renewTask(key, token, expireTime, renewPeriod);
} else {
log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}", key);
}
} catch (Exception e) {
// 這里異常是拋不出去的，所以需要 catch 打印
log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}", key, e);
}
} else {
log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，不再持有token，key : {}", key);
}
}, renewPeriod, TimeUnit.NANOSECONDS);
}
}這就是一個(gè)最簡單的實(shí)現(xiàn)方式。不過這里存在著許多問題：
- 續(xù)約任務(wù)
這里判斷是否持有令牌和續(xù)約這兩個(gè)動(dòng)作不在同一個(gè)事務(wù)里，可能發(fā)生覆蓋現(xiàn)象。假設(shè)A線程判斷自己持有令牌，但是一直沒有請求 Redis 導(dǎo)致鎖過期。B線程成功獲鎖，這時(shí)A線程往下執(zhí)行 Redis 請求，結(jié)果A線程搶了B線程的鎖。
- 釋放鎖
這里判斷是否持有令牌和刪除key這兩個(gè)動(dòng)作不在同一個(gè)事務(wù)里，可能出現(xiàn)誤刪現(xiàn)象。假設(shè)A線程現(xiàn)在要釋放鎖，通過了令牌判斷，準(zhǔn)備刪除 key 但是還沒執(zhí)行。這時(shí) key 過期了，B線程成功獲鎖。接著A線程執(zhí)行刪除 key 導(dǎo)致了 B 線程的鎖被刪除。
因此，判斷持有令牌與續(xù)約/刪除key這兩個(gè)動(dòng)作是需要原子性的，我們可以通過 lua 來實(shí)現(xiàn)。
擴(kuò)展，了解管道與 lua 的區(qū)別
- pipeline（多用于命令簡單高效，無關(guān)聯(lián)的場景）
優(yōu)點(diǎn)：使用簡單，有效減少網(wǎng)絡(luò)IO
缺點(diǎn)：本質(zhì)還是發(fā)送命令請求Redis 服務(wù)，如果效率過低，就會(huì)阻塞 Redis，導(dǎo)致 Redis 無法處理其他請求
- lua（多用于命令復(fù)雜，命令間有關(guān)聯(lián)的場景）
優(yōu)點(diǎn)：
- Redis 支持 lua 腳本，Redis 服務(wù)執(zhí)行 lua 的同時(shí)是可以處理別的請求的，不會(huì)產(chǎn)生阻塞
- 命令都在腳本中，有效減少網(wǎng)絡(luò)IO
- 具有原子性
缺點(diǎn)：
有一定的學(xué)習(xí)成本
3.1.1 使用 lua 進(jìn)行優(yōu)化
RedisLockUtil 代碼如下:
package com.example.demo.utils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author CaiZhuliang
* @date 2023/8/31
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisLockUtil {
private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
new BasicThreadFactory.Builder()
.namingPattern("redisLockUtil-schedule-pool-%d")
.daemon(true)
.build());
private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<>();
private static final String SUCCESS = "1";
/**
* 允許當(dāng)前token續(xù)約
*/
private static final Integer CAN_RENEW = 0;
/**
* 記錄token的狀態(tài)，0-可以續(xù)約，其他情況均不能續(xù)約
*/
private static final Map<String, Integer> TOKEN_STATUS = Maps.newConcurrentMap();
private final RedisTemplate<String, String> redisTemplate;
/**
* 釋放鎖，這個(gè)方法與 com.example.demo.utils.RedisLockUtil#simpleLock(java.lang.String, java.lang.Long) 配對。
* <p>必須和RedisLockUtil#simpleLock是同一個(gè)線程</p>
* @param key 需要釋放鎖的key
* @return true-成功 false-失敗
*/
public boolean releaseSimpleLock(String key) {
String token = THREAD_LOCAL.get();
if (null != token) {
TOKEN_STATUS.put(token, 1);
}
try {
String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
"then redis.call('expire', KEYS[1], 0) return '1' end " +
"return '0'";
DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token);
log.info("非cluster模式簡單分布式鎖 - 釋放key: {}, result : {}, token : {}", key, result, token);
return SUCCESS.equals(result);
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}", key, e);
return false;
} finally {
THREAD_LOCAL.remove();
if (null != token) {
TOKEN_STATUS.remove(token);
}
}
}
/**
* 簡單分布式鎖實(shí)現(xiàn)，續(xù)約周期是 expireTime 的一半。舉個(gè)例子， expireTime = 8000，那么鎖續(xù)約將會(huì)是每 4000 毫秒續(xù)約一次
* <p>這個(gè)方法不考慮Redis的集群架構(gòu)，不考慮腦裂問題，當(dāng)只有一個(gè) Redis來考慮。</p>
* <p>這個(gè)方法使用 com.example.demo.utils.RedisLockUtil#releaseSimpleLock(java.lang.String) 來釋放鎖</p>
* @param key 需要上鎖的key
* @param expireTime 過期時(shí)間，單位：毫秒
* @return true-成功 false-失敗
*/
public boolean simpleLock(String key, Long expireTime) {
if (StringUtils.isBlank(key)) {
log.warn("非cluster模式簡單分布式鎖 - key is blank");
return false;
}
if (null == expireTime || expireTime <= 0) {
expireTime = 0L;
}
// 續(xù)約周期，單位納秒
long renewPeriod = expireTime / 2 * 1000_000;
try {
String token = System.currentTimeMillis() + ":" + UUID.randomUUID();
// 設(shè)置鎖
Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
if (Boolean.FALSE.equals(result)) {
return false;
}
log.info("非cluster模式簡單分布式鎖 - 上鎖成功，key : {}, token : {}", key, token);
// 上鎖成功后將令牌綁定當(dāng)前線程
THREAD_LOCAL.set(token);
TOKEN_STATUS.put(token, 0);
if (renewPeriod > 0) {
// 續(xù)約任務(wù)
renewTask(key, token, expireTime, renewPeriod);
}
return true;
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 上鎖發(fā)生異常，key : {}", key, e);
String token = THREAD_LOCAL.get();
if (StringUtils.isNotBlank(token)) {
if (!releaseSimpleLock(key)) {
log.warn("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生失敗，key : {}, token : {}", key, token);
}
}
return false;
}
}
/**
* 鎖續(xù)約任務(wù)
* @param key 需要續(xù)命的key
* @param token 成功獲鎖的線程持有的令牌
* @param expireTime 過期時(shí)間，單位：毫秒
* @param renewPeriod 續(xù)約周期，單位：納秒
*/
private void renewTask(String key, String token, long expireTime, long renewPeriod) {
if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
EXECUTOR_SERVICE.schedule(() -> {
if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
try {
String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
"then " +
" if (redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])) " +
" then return '1' else return redis.call('get', KEYS[1]) end " +
"end " +
"return redis.call('get', KEYS[1])";
DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token, String.valueOf(expireTime));
if (SUCCESS.equals(result)) {
// 續(xù)約成功
log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}", key);
// 開啟下一次續(xù)約任務(wù)
renewTask(key, token, expireTime, renewPeriod);
} else {
// 打印下 result，看下是否因?yàn)椴辉俪钟辛钆茖?dǎo)致的續(xù)約失敗
log.warn("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}, token : {}, result : {}", key, token, result);
}
} catch (Exception e) {
// 這里異常是拋不出去的，所以需要 catch 打印
log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}", key, e);
}
}
}, renewPeriod, TimeUnit.NANOSECONDS);
}
}
}這里還有一個(gè)問題：如果redis.call('get', KEYS[1]) == ARGV[1] 成立，但是執(zhí)行redis.call('expire', KEYS[1], 0) 失敗，怎么辦？我這里已經(jīng)執(zhí)行了THREAD_LOCAL.remove()，想重復(fù)釋放是不可能的了，但是我這里不能不 remove 或者僅當(dāng) Redis 釋放鎖成功才 remove，這樣存在內(nèi)存泄漏的風(fēng)險(xiǎn)。要怎么處理呢？
這是優(yōu)化后的代碼：
package com.example.demo.utils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author CaiZhuliang
* @date 2023/8/31
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisSimpleLockUtil {
private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
new BasicThreadFactory.Builder()
.namingPattern("redisSimpleLockUtil-schedule-pool-%d")
.daemon(true)
.build());
private static final ThreadLocal<String> THREAD_LOCAL_TOKEN = new ThreadLocal<>();
private static final String SUCCESS = "1";
/**
* 允許當(dāng)前token續(xù)約
*/
private static final Integer CAN_RENEW = 0;
/**
* 記錄token的狀態(tài)，0-可以續(xù)約，其他情況均不能續(xù)約
*/
private static final Map<String, Integer> TOKEN_STATUS = Maps.newConcurrentMap();
private final RedisTemplate<String, String> redisTemplate;
/**
* 釋放鎖
* <p>必須和 RedisSimpleLockUtil#lock 是同一個(gè)線程</p>
* @param key key 需要釋放鎖的key
* @param token 持有的令牌
* @return true-成功 false-失敗
*/
public boolean releaseLock(String key, String token) {
if (StringUtils.isBlank(token)) {
return false;
}
TOKEN_STATUS.put(token, 1);
try {
String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
"then redis.call('expire', KEYS[1], 0) return '1' end " +
"return '0'";
DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token);
log.info("非cluster模式簡單分布式鎖 - 釋放key: {}, result : {}, token : {}", key, result, token);
if (SUCCESS.equals(result)) {
return true;
}
String remoteToken = redisTemplate.opsForValue().get(key);
if (token.equals(remoteToken)) {
log.warn("非cluster模式簡單分布式鎖 - 釋放鎖失敗，key : {}, token : {}", key, token);
return false;
}
return true;
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}, token : {}", key, token, e);
return false;
} finally {
THREAD_LOCAL_TOKEN.remove();
TOKEN_STATUS.remove(token);
}
}
/**
* 簡單分布式鎖實(shí)現(xiàn)，續(xù)約周期是 expireTime 的一半。舉個(gè)例子， expireTime = 8000，那么鎖續(xù)約將會(huì)是每 4000 毫秒續(xù)約一次
* <p>這個(gè)方法不考慮Redis的集群架構(gòu)，不考慮腦裂問題，當(dāng)只有一個(gè)Redis來考慮。</p>
* @param key 需要上鎖的key
* @param expireTime 過期時(shí)間，單位：毫秒
* @return 上鎖成功返回令牌，失敗則返回空串
*/
public String lock(String key, Long expireTime) {
if (StringUtils.isBlank(key)) {
log.warn("非cluster模式簡單分布式鎖 - key is blank");
return StringUtils.EMPTY;
}
if (null == expireTime || expireTime <= 0) {
expireTime = 0L;
}
// 續(xù)約周期，單位納秒
long renewPeriod = expireTime * 500_000;
try {
String token = System.currentTimeMillis() + ":" + UUID.randomUUID();
// 設(shè)置鎖
Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
if (Boolean.FALSE.equals(result)) {
return StringUtils.EMPTY;
}
log.info("非cluster模式簡單分布式鎖 - 上鎖成功，key : {}, token : {}", key, token);
// 上鎖成功后將令牌綁定當(dāng)前線程
THREAD_LOCAL_TOKEN.set(token);
TOKEN_STATUS.put(token, 0);
if (renewPeriod > 0) {
// 續(xù)約任務(wù)
log.info("非cluster模式簡單分布式鎖 - 添加續(xù)約任務(wù)，key : {}, token : {}, renewPeriod : {}納秒", key, token, renewPeriod);
renewTask(key, token, expireTime, renewPeriod);
}
return token;
} catch (Exception e) {
String token = THREAD_LOCAL_TOKEN.get();
log.error("非cluster模式簡單分布式鎖 - 上鎖發(fā)生異常，key : {}, token : {}", key, token, e);
return StringUtils.isBlank(token) ? StringUtils.EMPTY : token;
}
}
/**
* 鎖續(xù)約任務(wù)
* @param key 需要續(xù)命的key
* @param token 成功獲鎖的線程持有的令牌
* @param expireTime 過期時(shí)間，單位：毫秒
* @param renewPeriod 續(xù)約周期，單位：納秒
*/
private void renewTask(String key, String token, long expireTime, long renewPeriod) {
try {
EXECUTOR_SERVICE.schedule(() -> {
try {
String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
"then " +
" if (redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])) " +
" then return '1' else return redis.call('get', KEYS[1]) end " +
"end " +
"return redis.call('get', KEYS[1])";
DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token, String.valueOf(expireTime));
if (SUCCESS.equals(result)) {
// 續(xù)約成功
log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}, token : {}", key, token);
// 這里加判斷是為了減少定時(shí)任務(wù)
if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
// 開啟下一次續(xù)約任務(wù)
renewTask(key, token, expireTime, renewPeriod);
}
} else {
// 這里加判斷是為了防止誤打印warn日志
if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
log.warn("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}, token : {}, result : {}", key, token, result);
}
}
} catch (Exception e) {
// 這里異常是拋不出去的，所以需要 catch 打印
log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}, token : {}", key, token, e);
}
}, renewPeriod, TimeUnit.NANOSECONDS);
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 添加鎖續(xù)約任務(wù)發(fā)生異常，key : {}, token : {}", key, token, e);
}
}
}package com.example.demo.utils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author CaiZhuliang
* @date 2023/8/31
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisSimpleLockUtil {
private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
new BasicThreadFactory.Builder()
.namingPattern("redisSimpleLockUtil-schedule-pool-%d")
.daemon(true)
.build());
private static final ThreadLocal<String> THREAD_LOCAL_TOKEN = new ThreadLocal<>();
private static final String SUCCESS = "1";
/**
* 允許當(dāng)前token續(xù)約
*/
private static final Integer CAN_RENEW = 0;
/**
* 記錄token的狀態(tài)，0-可以續(xù)約，其他情況均不能續(xù)約
*/
private static final Map<String, Integer> TOKEN_STATUS = Maps.newConcurrentMap();
private final RedisTemplate<String, String> redisTemplate;
/**
* 釋放鎖
* <p>必須和 RedisSimpleLockUtil#lock 是同一個(gè)線程</p>
* @param key key 需要釋放鎖的key
* @param token 持有的令牌
* @return true-成功 false-失敗
*/
public boolean releaseLock(String key, String token) {
if (StringUtils.isBlank(token)) {
return false;
}
TOKEN_STATUS.put(token, 1);
try {
String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
"then redis.call('expire', KEYS[1], 0) return '1' end " +
"return '0'";
DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token);
log.info("非cluster模式簡單分布式鎖 - 釋放key: {}, result : {}, token : {}", key, result, token);
if (SUCCESS.equals(result)) {
return true;
}
String remoteToken = redisTemplate.opsForValue().get(key);
if (token.equals(remoteToken)) {
log.warn("非cluster模式簡單分布式鎖 - 釋放鎖失敗，key : {}, token : {}", key, token);
return false;
}
return true;
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}, token : {}", key, token, e);
return false;
} finally {
THREAD_LOCAL_TOKEN.remove();
TOKEN_STATUS.remove(token);
}
}
/**
* 簡單分布式鎖實(shí)現(xiàn)，續(xù)約周期是 expireTime 的一半。舉個(gè)例子， expireTime = 8000，那么鎖續(xù)約將會(huì)是每 4000 毫秒續(xù)約一次
* <p>這個(gè)方法不考慮Redis的集群架構(gòu)，不考慮腦裂問題，當(dāng)只有一個(gè)Redis來考慮。</p>
* @param key 需要上鎖的key
* @param expireTime 過期時(shí)間，單位：毫秒
* @return 上鎖成功返回令牌，失敗則返回空串
*/
public String lock(String key, Long expireTime) {
if (StringUtils.isBlank(key)) {
log.warn("非cluster模式簡單分布式鎖 - key is blank");
return StringUtils.EMPTY;
}
if (null == expireTime || expireTime <= 0) {
expireTime = 0L;
}
// 續(xù)約周期，單位納秒
long renewPeriod = expireTime * 500_000;
try {
String token = System.currentTimeMillis() + ":" + UUID.randomUUID();
// 設(shè)置鎖
Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
if (Boolean.FALSE.equals(result)) {
return StringUtils.EMPTY;
}
log.info("非cluster模式簡單分布式鎖 - 上鎖成功，key : {}, token : {}", key, token);
// 上鎖成功后將令牌綁定當(dāng)前線程
THREAD_LOCAL_TOKEN.set(token);
TOKEN_STATUS.put(token, 0);
if (renewPeriod > 0) {
// 續(xù)約任務(wù)
log.info("非cluster模式簡單分布式鎖 - 添加續(xù)約任務(wù)，key : {}, token : {}, renewPeriod : {}納秒", key, token, renewPeriod);
renewTask(key, token, expireTime, renewPeriod);
}
return token;
} catch (Exception e) {
String token = THREAD_LOCAL_TOKEN.get();
log.error("非cluster模式簡單分布式鎖 - 上鎖發(fā)生異常，key : {}, token : {}", key, token, e);
return StringUtils.isBlank(token) ? StringUtils.EMPTY : token;
}
}
/**
* 鎖續(xù)約任務(wù)
* @param key 需要續(xù)命的key
* @param token 成功獲鎖的線程持有的令牌
* @param expireTime 過期時(shí)間，單位：毫秒
* @param renewPeriod 續(xù)約周期，單位：納秒
*/
private void renewTask(String key, String token, long expireTime, long renewPeriod) {
try {
EXECUTOR_SERVICE.schedule(() -> {
try {
String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
"then " +
" if (redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])) " +
" then return '1' else return redis.call('get', KEYS[1]) end " +
"end " +
"return redis.call('get', KEYS[1])";
DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token, String.valueOf(expireTime));
if (SUCCESS.equals(result)) {
// 續(xù)約成功
log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}, token : {}", key, token);
// 這里加判斷是為了減少定時(shí)任務(wù)
if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
// 開啟下一次續(xù)約任務(wù)
renewTask(key, token, expireTime, renewPeriod);
}
} else {
// 這里加判斷是為了防止誤打印warn日志
if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
log.warn("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}, token : {}, result : {}", key, token, result);
}
}
} catch (Exception e) {
// 這里異常是拋不出去的，所以需要 catch 打印
log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}, token : {}", key, token, e);
}
}, renewPeriod, TimeUnit.NANOSECONDS);
} catch (Exception e) {
log.error("非cluster模式簡單分布式鎖 - 添加鎖續(xù)約任務(wù)發(fā)生異常，key : {}, token : {}", key, token, e);
}
}
}下面是并發(fā)單元測試代碼：
@Test
public void concurrencyTest() {
String[] nums = {"1", "2", "3", "4", "5"};
List<CompletableFuture<Void>> list = Lists.newArrayListWithExpectedSize(100);
for (int i = 0; i < 50; i++) {
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
for (int count = 0; count < 10; count++) {
int random = new Random().nextInt(100) % 5;
String key = "test_" + nums[random];
while (true) {
String token = redisSimpleLockUtil.lock(key, 3_000L);
if (StringUtils.isNotBlank(token)) {
log.info("concurrencyTest - key : {}", key);
try {
Thread.sleep(new Random().nextInt(1500));
} catch (Exception e) {
log.error("concurrencyTest - 發(fā)生異常, key : {}", key, e);
} finally {
boolean unlock = redisSimpleLockUtil.releaseLock(key, token);
if (!unlock) {
log.error("concurrencyTest - 釋放鎖失敗，key : {}", key);
}
}
break;
}
}
}
});
list.add(future);
}
CompletableFuture<?>[] futures = new CompletableFuture[list.size()];
list.toArray(futures);
CompletableFuture.allOf(futures).join();
}3.2 紅鎖
一般公司使用Redis 時(shí)都不可能是單節(jié)點(diǎn)的，要么主從+哨兵架構(gòu)，要么就是 cluster 架構(gòu)。面對集群，我們不得不思考如何應(yīng)對腦裂這個(gè)問題。而 Redlock 是Redis官方網(wǎng)站給出的解決方案。
下面看下針對這兩種集群架構(gòu)的處理方式：
- 主從+哨兵
通過訪問哨兵獲取當(dāng)前 master 節(jié)點(diǎn)，統(tǒng)計(jì)票數(shù)，超過半數(shù)的 master 節(jié)點(diǎn)就是真的 master。我們可以對比我們成功上鎖的節(jié)點(diǎn)是否是真的 master node，從而避免腦裂問題。
- cluster
- 上鎖需要在集群中半數(shù)以上的 master 操作成功了才算成功。
3.2.1 紅鎖的問題
紅鎖通過過半原則來規(guī)避腦裂，但是這就讓我們不得不考慮訪問節(jié)點(diǎn)的等待超時(shí)時(shí)間應(yīng)該要多長。而且，也會(huì)降低Redis 分布式鎖的吞吐量。如果有半數(shù)節(jié)點(diǎn)不可用，那么分布式鎖也將變得不可用。因此，實(shí)際使用中，我們還要結(jié)合自己實(shí)際的業(yè)務(wù)場景來權(quán)衡要不要用紅鎖或者修改實(shí)現(xiàn)方案。
作者介紹
蔡柱梁，51CTO社區(qū)編輯，從事Java后端開發(fā)8年，做過傳統(tǒng)項(xiàng)目廣電BOSS系統(tǒng)，后投身互聯(lián)網(wǎng)電商，負(fù)責(zé)過訂單，TMS，中間件等。
責(zé)任編輯：華軒
來源：
51CTO
