作者 | 蔡柱梁
審校 | 重樓
目錄
- 分布式鎖介紹
 - 如何實(shí)現(xiàn)分布式鎖
 - 實(shí)現(xiàn)分布式鎖
 
1 分布式鎖介紹
現(xiàn)在的服務(wù)往往都是多節(jié)點(diǎn),在一些特定的場景下容易產(chǎn)生并發(fā)問題,比如扣減庫存,送完即止活動,中臺的批量導(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è)動作不在同一個(gè)事務(wù)里,可能發(fā)生覆蓋現(xiàn)象。假設(shè)A線程判斷自己持有令牌,但是一直沒有請求 Redis 導(dǎo)致鎖過期。B線程成功獲鎖,這時(shí)A線程往下執(zhí)行 Redis 請求,結(jié)果A線程搶了B線程的鎖。
- 釋放鎖
 
這里判斷是否持有令牌和刪除key這兩個(gè)動作不在同一個(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è)動作是需要原子性的,我們可以通過 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ù),如果效率過低,就會阻塞 Redis,導(dǎo)致 Redis 無法處理其他請求
- lua(多用于命令復(fù)雜,命令間有關(guān)聯(lián)的場景)
 
優(yōu)點(diǎn):
- Redis 支持 lua 腳本,Redis 服務(wù)執(zhí)行 lua 的同時(shí)是可以處理別的請求的,不會產(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ù)約將會是每 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ù)約將會是每 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ù)約將會是每 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)該要多長。而且,也會降低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,中間件等。















 
 
 







 
 
 
 