SpringBoot整合Redis实战：从缓存穿透到分布式锁的完整解决方案

一、引言

在现代分布式系统架构中，缓存技术已成为提升系统性能不可或缺的组成部分。Spring Boot作为当前最流行的Java应用开发框架，以其"约定优于配置"的理念极大地简化了Spring应用的初始搭建和开发过程。而Redis作为高性能的内存数据结构存储系统，凭借其丰富的数据类型、原子性操作和出色的性能，成为分布式缓存的首选解决方案。

然而，在实际生产环境中，开发者往往会面临缓存穿透、缓存雪崩、缓存击穿等典型问题，这些问题如果处理不当，轻则导致系统性能下降，重则引发服务不可用。同时，在分布式环境下，如何保证数据一致性也成为系统设计的关键挑战，这时分布式锁便成为解决问题的利器。

本文将全面探讨Spring Boot与Redis的整合实践，从基础配置到高级应用，重点分析缓存穿透问题的多种解决方案，并深入讲解基于Redis的分布式锁实现机制。通过理论结合实践的方式，为开发者提供一套完整的Redis应用方案，帮助构建高性能、高可用的分布式系统。

二、Spring Boot整合Redis基础配置

2.1 环境准备与依赖配置

在开始整合之前，需要确保已具备以下环境条件：

• JDK 1.8或更高版本
• Maven 3.x或Gradle构建工具
• Redis服务器(建议3.2以上版本)
• Spring Boot 2.x(本文以2.5.4为例)

在Spring Boot项目中集成Redis非常简单，只需在pom.xml中添加相关依赖：

<dependencies>
<!– Spring Data Redis –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!– 连接池依赖 –>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>

<!– 可选：Redisson分布式锁 –>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.16.0</version>
</dependency>
</dependencies>

值得注意的是，从Spring Boot 2.x开始，默认使用Lettuce作为Redis客户端，而非早期的Jedis。Lettuce基于Netty实现，支持异步和响应式编程模型，在连接管理和高并发场景下表现更优。

2.2 Redis连接配置

在application.yml(或application.properties)中配置Redis连接信息：

spring:
redis:
host: 127.0.0.1
port: 6379
password: yourpassword # 若无密码可省略
database: 0
lettuce:
pool:
max-active: 20 # 连接池最大连接数
max-idle: 10 # 连接池最大空闲连接数
min-idle: 5 # 连接池最小空闲连接数
max-wait: 3000ms # 获取连接最大等待时间

2.3 RedisTemplate配置与使用

Spring Data Redis提供了RedisTemplate和StringRedisTemplate两个模板类来简化Redis操作。为了支持更丰富的数据类型和序列化方式，通常需要自定义配置：

@Configuration
public class RedisConfig {

@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(factory);

// 使用Jackson2JsonRedisSerializer序列化value
Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<>(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.activateDefaultTyping(om.getPolymorphicTypeValidator(),
ObjectMapper.DefaultTyping.NON_FINAL);
serializer.setObjectMapper(om);

// 设置key和value的序列化规则
template.setKeySerializer(new StringRedisSerializer());
template.setValueSerializer(serializer);
template.setHashKeySerializer(new StringRedisSerializer());
template.setHashValueSerializer(serializer);

template.afterPropertiesSet();
return template;
}
}

配置完成后，即可在Service中注入RedisTemplate进行操作：

@Service
public class CacheService {

@Autowired
private RedisTemplate<String, Object> redisTemplate;

public void setValue(String key, Object value, long timeout, TimeUnit unit) {
redisTemplate.opsForValue().set(key, value, timeout, unit);
}

public Object getValue(String key) {
return redisTemplate.opsForValue().get(key);
}

// 其他操作…
}

三、Redis缓存实战与性能优化

3.1 Spring Cache抽象与Redis集成

Spring框架提供了强大的缓存抽象，可以非常方便地与Redis集成。首先需要在启动类上添加@EnableCaching注解：

@SpringBootApplication
@EnableCaching
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}

然后配置Redis缓存管理器：

@Configuration
public class CacheConfig {

@Bean
public RedisCacheManager cacheManager(RedisConnectionFactory factory) {
return RedisCacheManager.builder(factory)
.cacheDefaults(RedisCacheConfiguration.defaultCacheConfig()
.entryTtl(Duration.ofMinutes(30)) // 默认缓存30分钟
.disableCachingNullValues()) // 不缓存null值
.withInitialCacheConfigurations(Collections.singletonMap(
"productCache", RedisCacheConfiguration.defaultCacheConfig()
.entryTtl(Duration.ofHours(2)) // 商品缓存2小时
.serializeValuesWith(RedisSerializationContext.SerializationPair
.fromSerializer(new GenericJackson2JsonRedisSerializer()))
))
.transactionAware()
.build();
}
}

在业务方法上使用缓存注解：

@Service
public class ProductService {

@Cacheable(value = "productCache", key = "#id")
public Product getProductById(Long id) {
// 模拟数据库查询
return productRepository.findById(id).orElse(null);
}

@CachePut(value = "productCache", key = "#product.id")
public Product updateProduct(Product product) {
return productRepository.save(product);
}

@CacheEvict(value = "productCache", key = "#id")
public void deleteProduct(Long id) {
productRepository.deleteById(id);
}
}

3.2 缓存穿透问题与解决方案

缓存穿透是指查询一个不存在的数据，由于缓存中查不到，每次请求都会穿透到数据库，导致数据库压力过大。解决方案如下：

方案一：缓存空对象

public Product getProductWithNullCache(Long id) {
// 先从缓存获取
Product product = (Product) redisTemplate.opsForValue().get("product:" + id);

if (product != null) {
// 特殊标记的空对象
if (product.getId() == null) {
return null; // 空对象直接返回null
}
return product;
}

// 缓存不存在，查询数据库
product = productRepository.findById(id).orElse(null);

if (product == null) {
// 数据库不存在，缓存一个特殊空对象，设置较短过期时间
Product nullProduct = new Product(); // 无id的特殊对象
redisTemplate.opsForValue().set("product:" + id, nullProduct, 5, TimeUnit.MINUTES);
return null;
}

// 数据库存在，写入缓存
redisTemplate.opsForValue().set("product:" + id, product, 2, TimeUnit.HOURS);
return product;
}

方案二：布隆过滤器

布隆过滤器是一种空间效率极高的概率型数据结构，用于判断一个元素是否存在于集合中。

@PostConstruct
public void initBloomFilter() {
List<Long> allIds = productRepository.findAllIds();
for (Long id : allIds) {
bloomFilter.put(id);
}
}

public Product getProductWithBloomFilter(Long id) {
if (!bloomFilter.mightContain(id)) {
return null; // 肯定不存在
}

// 可能存在，继续正常缓存查询流程
return getProductById(id);
}

3.3 缓存雪崩问题与解决方案

缓存雪崩是指缓存中大量数据同时过期，导致所有请求都落到数据库上，造成数据库瞬时压力过大甚至崩溃。

解决方案：

// 在设置缓存时，基础过期时间加上随机值
public void setWithRandomExpire(String key, Object value, long baseTimeout, TimeUnit unit) {
// 随机增加0-30分钟的偏移量
long randomOffset = (long) (Math.random() * 30 * 60 * 1000);
long timeout = unit.toMillis(baseTimeout) + randomOffset;
redisTemplate.opsForValue().set(key, value, timeout, TimeUnit.MILLISECONDS);
}

// 使用Caffeine作为本地缓存
@Bean
public Cache<Long, Product> localCache() {
return Caffeine.newBuilder()
.maximumSize(1000)
.expireAfterWrite(10, TimeUnit.MINUTES)
.build();
}

public Product getProductWithMultiLevelCache(Long id) {
// 1. 先查本地缓存
Product product = localCache.getIfPresent(id);
if (product != null) {
return product;
}

// 2. 查Redis缓存
product = (Product) redisTemplate.opsForValue().get("product:" + id);
if (product != null) {
localCache.put(id, product); // 回填本地缓存
return product;
}

// 3. 查数据库
product = productRepository.findById(id).orElse(null);
if (product != null) {
redisTemplate.opsForValue().set("product:" + id, product, 2, TimeUnit.HOURS);
localCache.put(id, product);
}

return product;
}

@Component
public class CachePreheat implements CommandLineRunner {

@Autowired
private ProductRepository productRepository;

@Autowired
private RedisTemplate<String, Object> redisTemplate;

@Override
public void run(String... args) {
// 加载前100个热门商品
List<Product> hotProducts = productRepository.findTop100ByOrderBySalesDesc();
hotProducts.forEach(product -> {
redisTemplate.opsForValue().set(
"product:" + product.getId(),
product,
1 + (long)(Math.random() * 3), // 1-4小时随机过期
TimeUnit.HOURS
);
});
}
}

3.4 缓存击穿问题与解决方案

缓存击穿是指某个热点key过期时，大量并发请求同时穿透到数据库，导致数据库压力激增。

解决方案：

public Product getProductWithMutexLock(Long id) {
String cacheKey = "product:" + id;
String lockKey = "lock:product:" + id;

// 1. 先查缓存
Product product = (Product) redisTemplate.opsForValue().get(cacheKey);
if (product != null) {
return product;
}

// 2. 尝试获取锁
String uuid = UUID.randomUUID().toString();
boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, uuid, 30, TimeUnit.SECONDS);

if (locked) {
try {
// 3. 再次检查缓存(可能在等待锁期间已被其他线程填充)
product = (Product) redisTemplate.opsForValue().get(cacheKey);
if (product != null) {
return product;
}

// 4. 查询数据库
product = productRepository.findById(id).orElse(null);
if (product != null) {
redisTemplate.opsForValue().set(cacheKey, product, 2, TimeUnit.HOURS);
} else {
// 缓存空对象防止穿透
redisTemplate.opsForValue().set(cacheKey, new Product(), 5, TimeUnit.MINUTES);
}

return product;
} finally {
// 释放锁 – 使用Lua脚本保证原子性
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
redisTemplate.execute(new DefaultRedisScript<>(script, Long.class),
Collections.singletonList(lockKey), uuid);
}
} else {
// 未获取到锁，短暂休眠后重试
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return getProductWithMutexLock(id); // 递归调用
}
}

@Data
public class RedisData<T> implements Serializable {
private T data;
private long expireTime; // 逻辑过期时间戳
}

public Product getProductWithLogicalExpire(Long id) {
String cacheKey = "product:" + id;

// 1. 从缓存获取数据
RedisData<Product> redisData = (RedisData<Product>) redisTemplate.opsForValue().get(cacheKey);
Product product = redisData.getData();

// 2. 判断是否逻辑过期
if (redisData.getExpireTime() > System.currentTimeMillis()) {
// 未过期，直接返回
return product;
}

// 3. 已过期，尝试获取锁重建缓存
String lockKey = "lock:product:" + id;
boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 30, TimeUnit.SECONDS);

if (locked) {
try {
// 4. 再次检查(可能在等待锁期间已被其他线程更新)
redisData = (RedisData<Product>) redisTemplate.opsForValue().get(cacheKey);
if (redisData.getExpireTime() > System.currentTimeMillis()) {
return redisData.getData();
}

// 5. 查询数据库
product = productRepository.findById(id).orElse(null);
if (product != null) {
// 设置新的逻辑过期时间(当前时间+2小时)
RedisData<Product> newRedisData = new RedisData<>();
newRedisData.setData(product);
newRedisData.setExpireTime(System.currentTimeMillis() + TimeUnit.HOURS.toMillis(2));
redisTemplate.opsForValue().set(cacheKey, newRedisData);
}

return product;
} finally {
redisTemplate.delete(lockKey);
}
} else {
// 未获取到锁，返回旧数据
return product;
}
}

四、Redis分布式锁深度解析

4.1 分布式锁核心要求

一个可靠的分布式锁应满足以下基本要求：

4.2 基于SETNX的分布式锁实现

Redis的SETNX(SET if Not eXists)命令是实现分布式锁的基础，但单纯使用SETNX会存在一些问题：

// 简单但不完善的实现
public boolean tryLock(String lockKey, String clientId, long expireTime) {
return redisTemplate.opsForValue().setIfAbsent(lockKey, clientId, expireTime, TimeUnit.SECONDS);
}

public void unlock(String lockKey, String clientId) {
if (clientId.equals(redisTemplate.opsForValue().get(lockKey))) {
redisTemplate.delete(lockKey);
}
}

这种实现存在以下问题：

4.3 完善的分布式锁实现

以下是改进后的分布式锁实现：

@Component
public class RedisDistributedLock {

@Autowired
private RedisTemplate<String, String> redisTemplate;

private static final String UNLOCK_SCRIPT =
"if redis.call('get', KEYS[1]) == ARGV[1] then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end";

public boolean tryLock(String lockKey, String clientId, long expireTime) {
return redisTemplate.opsForValue()
.setIfAbsent(lockKey, clientId, expireTime, TimeUnit.SECONDS);
}

public boolean unlock(String lockKey, String clientId) {
DefaultRedisScript<Long> script = new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class);
Long result = redisTemplate.execute(script, Collections.singletonList(lockKey), clientId);
return result != null && result == 1;
}

public boolean lockWithRetry(String lockKey, String clientId,
long expireTime, int maxRetry, long waitTime) {
int retry = 0;
while (retry < maxRetry) {
if (tryLock(lockKey, clientId, expireTime)) {
return true;
}
try {
Thread.sleep(waitTime);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;

五、基于注解的Redis分布式锁实现

5.1 自定义分布式锁注解

要实现基于注解的分布式锁，首先需要定义一个自定义注解：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisLock {
String key(); // 锁的key
String prefix() default ""; // key前缀
long expire() default 30; // 锁的过期时间(秒)
long waitTime() default 10; // 获取锁的最大等待时间(秒)
TimeUnit timeUnit() default TimeUnit.SECONDS; // 时间单位
boolean retry() default true; // 获取锁失败是否重试
}

5.2 切面实现分布式锁逻辑

通过Spring AOP实现注解的切面处理：

@Aspect
@Component
@Slf4j
public class RedisLockAspect {

@Autowired
private RedisTemplate<String, String> redisTemplate;

private static final String LOCK_PREFIX = "lock:";
private static final String LOCK_VALUE = "locked";
private static final String UNLOCK_SCRIPT =
"if redis.call('get', KEYS[1]) == ARGV[1] then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end";

@Around("@annotation(redisLock)")
public Object around(ProceedingJoinPoint joinPoint, RedisLock redisLock) throws Throwable {
// 构造完整的锁key
String lockKey = LOCK_PREFIX + redisLock.prefix() + SpelUtils.parseKey(redisLock.key(), joinPoint);

// 获取锁
boolean locked = false;
try {
locked = tryLock(lockKey, redisLock.waitTime(), redisLock.expire(), redisLock.timeUnit());

if (!locked && !redisLock.retry()) {
throw new RuntimeException("获取分布式锁失败");
}

if (locked || (redisLock.retry() && tryLockWithRetry(lockKey,
redisLock.waitTime(), redisLock.expire(), redisLock.timeUnit()))) {
return joinPoint.proceed();
} else {
throw new RuntimeException("获取分布式锁失败，超过最大等待时间");
}
} finally {
if (locked) {
unlock(lockKey);
}
}
}

private boolean tryLock(String key, long waitTime, long expire, TimeUnit unit) {
long start = System.currentTimeMillis();
try {
// 尝试获取锁
Boolean success = redisTemplate.opsForValue()
.setIfAbsent(key, LOCK_VALUE, expire, unit);

if (success != null && success) {
return true;
}

// 计算剩余等待时间
long remainTime = unit.toMillis(waitTime) – (System.currentTimeMillis() – start);
if (remainTime <= 0) {
return false;
}

// 短暂休眠后重试
Thread.sleep(Math.min(100, remainTime));
return tryLock(key, waitTime, expire, unit);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}

private boolean tryLockWithRetry(String key, long waitTime, long expire, TimeUnit unit) {
long endTime = System.currentTimeMillis() + unit.toMillis(waitTime);
while (System.currentTimeMillis() < endTime) {
if (redisTemplate.opsForValue().setIfAbsent(key, LOCK_VALUE, expire, unit)) {
return true;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
return false;
}

private void unlock(String key) {
DefaultRedisScript<Long> script = new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class);
Long result = redisTemplate.execute(script, Collections.singletonList(key), LOCK_VALUE);
if (result == null || result == 0) {
log.warn("释放分布式锁失败，锁可能已过期或已被其他线程释放: {}", key);
}
}
}

5.3 SpEL表达式解析工具

为了支持在注解key中使用SpEL表达式，需要实现一个解析工具类：

public class SpelUtils {

private static final ExpressionParser parser = new SpelExpressionParser();
private static final ParameterNameDiscoverer parameterNameDiscoverer = new DefaultParameterNameDiscoverer();

public static String parseKey(String key, ProceedingJoinPoint joinPoint) {
if (key == null || key.isEmpty()) {
throw new IllegalArgumentException("RedisLock key cannot be empty");
}

// 获取方法签名
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();

// 创建解析上下文
EvaluationContext context = new StandardEvaluationContext();

// 获取参数名和参数值
String[] paramNames = parameterNameDiscoverer.getParameterNames(method);
Object[] args = joinPoint.getArgs();

if (paramNames != null) {
for (int i = 0; i < paramNames.length; i++) {
context.setVariable(paramNames[i], args[i]);
}
}

// 解析表达式
try {
Expression expression = parser.parseExpression(key);
Object value = expression.getValue(context);
return value != null ? value.toString() : "";
} catch (Exception e) {
log.warn("SpEL表达式解析失败，将使用原始字符串: {}", key, e);
return key;
}
}
}

5.4 使用示例

在业务方法上使用自定义注解：

@Service
public class OrderService {

@RedisLock(key = "'order:' + #orderId", prefix = "create:", expire = 10)
public void createOrder(Long orderId) {
// 创建订单业务逻辑
}

@RedisLock(key = "'order:pay:' + #orderId", expire = 30, waitTime = 5)
public void payOrder(Long orderId) {
// 支付订单业务逻辑
}

@RedisLock(key = "'stock:reduce:' + #productId", expire = 60)
public void reduceStock(Long productId, int quantity) {
// 扣减库存业务逻辑
}
}

5.5 高级特性扩展

5.5.1 锁的可重入性

为了实现可重入锁，可以改造锁的value为线程标识+重入次数：

private boolean tryReentrantLock(String key, long expire, TimeUnit unit) {
String currentThreadId = getThreadIdentifier();
String currentValue = redisTemplate.opsForValue().get(key);

// 如果是当前线程持有的锁，增加重入次数
if (currentValue != null && currentValue.startsWith(currentThreadId + ":")) {
int count = Integer.parseInt(currentValue.split(":")[1]);
redisTemplate.opsForValue().set(key, currentThreadId + ":" + (count + 1), expire, unit);
return true;
}

// 尝试获取新锁
if (redisTemplate.opsForValue().setIfAbsent(key, currentThreadId + ":1", expire, unit)) {
return true;
}

return false;
}

private void unlockReentrant(String key) {
String currentThreadId = getThreadIdentifier();
String currentValue = redisTemplate.opsForValue().get(key);

if (currentValue != null && currentValue.startsWith(currentThreadId + ":")) {
int count = Integer.parseInt(currentValue.split(":")[1]);
if (count > 1) {
// 减少重入次数
redisTemplate.opsForValue().set(key, currentThreadId + ":" + (count – 1));
} else {
// 完全释放锁
redisTemplate.delete(key);
}
}
}

private String getThreadIdentifier() {
return Thread.currentThread().getId() + "@" + ManagementFactory.getRuntimeMXBean().getName();
}

5.5.2 锁自动续期（看门狗机制）

对于执行时间不确定的长任务，可以实现锁自动续期：

private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

private boolean tryLockWithWatchdog(String key, long expire, TimeUnit unit) {
if (redisTemplate.opsForValue().setIfAbsent(key, LOCK_VALUE, expire, unit)) {
// 启动看门狗定时任务
long interval = unit.toMillis(expire) / 3;
scheduler.scheduleAtFixedRate(() -> {
if (redisTemplate.opsForValue().get(key) != null) {
redisTemplate.expire(key, expire, unit);
}
}, interval, interval, TimeUnit.MILLISECONDS);
return true;
}
return false;
}

六、Redisson高级分布式锁

6.1 Redisson简介

Redisson是Redis官方推荐的Java客户端，提供了丰富的分布式对象和服务，包括分布式锁的实现。相比原生Redis命令实现的锁，Redisson提供了更多高级特性：

6.2 Redisson配置

首先配置Redisson客户端：

@Configuration
public class RedissonConfig {

@Value("${spring.redis.host}")
private String host;

@Value("${spring.redis.port}")
private String port;

@Value("${spring.redis.password}")
private String password;

@Bean(destroyMethod = "shutdown")
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer()
.setAddress("redis://" + host + ":" + port)
.setPassword(password)
.setDatabase(0)
.setConnectionPoolSize(64)
.setConnectionMinimumIdleSize(10)
.setIdleConnectionTimeout(10000)
.setConnectTimeout(10000)
.setTimeout(3000)
.setRetryAttempts(3)
.setRetryInterval(1500);
return Redisson.create(config);
}
}

6.3 基于Redisson的注解实现

可以基于Redisson实现更强大的分布式锁注解：

@Aspect
@Component
@Slf4j
public class RedissonLockAspect {

@Autowired
private RedissonClient redissonClient;

@Around("@annotation(redissonLock)")
public Object around(ProceedingJoinPoint joinPoint, RedissonLock redissonLock) throws Throwable {
String lockKey = redissonLock.prefix() + SpelUtils.parseKey(redissonLock.key(), joinPoint);
RLock lock = redissonClient.getLock(lockKey);

try {
boolean locked = false;
if (redissonLock.waitTime() > 0) {
locked = lock.tryLock(redissonLock.waitTime(), redissonLock.leaseTime(), redissonLock.timeUnit());
} else {
lock.lock(redissonLock.leaseTime(), redissonLock.timeUnit());
locked = true;
}

if (locked || (redissonLock.retry() && tryLockWithRetry(lock,
redissonLock.waitTime(), redissonLock.leaseTime(), redissonLock.timeUnit()))) {
return joinPoint.proceed();
} else {
throw new RuntimeException("获取分布式锁失败，超过最大等待时间");
}
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}

private boolean tryLockWithRetry(RLock lock, long waitTime, long leaseTime, TimeUnit unit)
throws InterruptedException {
long endTime = System.currentTimeMillis() + unit.toMillis(waitTime);
while (System.currentTimeMillis() < endTime) {
if (lock.tryLock(100, leaseTime, unit)) {
return true;
}
}
return false;
}
}

6.4 Redisson分布式锁最佳实践

七、分布式锁的测试与验证

7.1 单元测试

编写单元测试验证分布式锁的基本功能：

@SpringBootTest
@Slf4j
class DistributedLockTest {

@Autowired
private RedisTemplate<String, String> redisTemplate;

@Autowired
private RedissonClient redissonClient;

@Test
void testRedisLock() throws InterruptedException {
String lockKey = "test:redis:lock";
int[] counter = {0};
int threadCount = 10;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);

CountDownLatch latch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
executor.execute(() -> {
boolean locked = false;
try {
locked = redisTemplate.opsForValue()
.setIfAbsent(lockKey, "locked", 10, TimeUnit.SECONDS);
if (locked) {
counter[0]++;
log.info("Counter: {}", counter[0]);
}
} finally {
if (locked) {
redisTemplate.delete(lockKey);
}
latch.countDown();
}
});
}

latch.await();
assertEquals(1, counter[0]);
}

@Test
void testRedissonLock() throws InterruptedException {
String lockKey = "test:redisson:lock";
int[] counter = {0};
int threadCount = 10;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);

CountDownLatch latch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
executor.execute(() -> {
RLock lock = redissonClient.getLock(lockKey);
try {
if (lock.tryLock(1, 10, TimeUnit.SECONDS)) {
counter[0]++;
log.info("Counter: {}", counter[0]);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
latch.countDown();
}
});
}

latch.await();
assertEquals(1, counter[0]);
}
}

7.2 集成测试

模拟高并发场景测试分布式锁的有效性：

@SpringBootTest
@Slf4j
class HighConcurrentLockTest {

@Autowired
private OrderService orderService;

@Test
void testCreateOrderWithLock() throws InterruptedException {
int threadCount = 100;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
CountDownLatch latch = new CountDownLatch(threadCount);

long startTime = System.currentTimeMillis();
for (int i = 0; i < threadCount; i++) {
final long orderId = 1000 + i;
executor.execute(() -> {
try {
orderService.createOrder(orderId);
} finally {
latch.countDown();
}
});
}

latch.await();
long duration = System.currentTimeMillis() – startTime;
log.info("Total time: {}ms", duration);
}
}

7.3 性能测试

使用JMH进行分布式锁性能基准测试：

@State(Scope.Benchmark)
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@Warmup(iterations = 3, time = 1)
@Measurement(iterations = 5, time = 3)
@Fork(1)
@Threads(8)
public class LockBenchmark {

@Autowired
private RedisTemplate<String, String> redisTemplate;

private RLock redissonLock;

@Setup
public void setup() {
redissonLock = redissonClient.getLock("benchmark:lock");
}

@Benchmark
public void redisLock() {
boolean locked = false;
try {
locked = redisTemplate.opsForValue()
.setIfAbsent("benchmark:redis:lock", "locked", 10, TimeUnit.SECONDS);
if (locked) {
// 模拟业务操作
Blackhole.consumeCPU(1000);
}
} finally {
if (locked) {
redisTemplate.delete("benchmark:redis:lock");
}
}
}

@Benchmark
public void redissonLock() {
try {
if (redissonLock.tryLock(100, 10, TimeUnit.MILLISECONDS)) {
// 模拟业务操作
Blackhole.consumeCPU(1000);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
if (redissonLock.isHeldByCurrentThread()) {
redissonLock.unlock();
}
}
}
}

八、生产环境最佳实践

8.1 Redis高可用部署

8.2 锁的监控与告警

九、Redis分布式锁的进阶实现方案

9.1 RedLock红锁算法实现

针对单点Redis的风险，Redis官方提出RedLock算法：

public class RedLockExample {
private List<RedissonClient> clients;

public boolean tryLock(String lockName, long waitTime, long leaseTime) {
List<RLock> locks = new ArrayList<>();
try {
// 1. 获取当前毫秒时间戳
long startTime = System.currentTimeMillis();

// 2. 尝试在所有节点获取锁
for (RedissonClient client : clients) {
RLock lock = client.getLock(lockName);
if (lock.tryLock(waitTime, leaseTime, TimeUnit.MILLISECONDS)) {
locks.add(lock);
}
}

// 3. 计算获取锁耗时
long elapsed = System.currentTimeMillis() – startTime;

// 4. 验证是否获取成功（N/2+1个节点）
return locks.size() >= clients.size()/2 + 1
&& elapsed < leaseTime;
} catch (Exception e) {
unlockAll(locks);
return false;
}
}

private void unlockAll(List<RLock> locks) {
locks.forEach(lock -> {
try { if (lock.isHeldByCurrentThread()) lock.unlock(); }
catch (Exception ignored) {}
});
}
}

9.2 锁分段优化技术

针对热点key的并发瓶颈，可采用分段锁策略：

public class SegmentLockService {
private static final int SEGMENTS = 16;
private final RedisLock[] locks;

public SegmentLockService() {
locks = new RedisLock[SEGMENTS];
for (int i = 0; i < SEGMENTS; i++) {
locks[i] = new RedisLock("segment_lock:" + i);
}
}

public void execute(String businessKey, Runnable task) {
int segment = Math.abs(businessKey.hashCode()) % SEGMENTS;
locks[segment].lock();
try {
task.run();
} finally {
locks[segment].unlock();
}
}
}

十、生产环境问题诊断方案

10.1 锁竞争监控指标体系

建议监控以下关键指标：

10.2 常见故障处理预案

十一、性能优化专项

11.1 网络优化方案

11.2 内存优化技巧

十二、多语言实现方案

12.1 Python实现

import redis
import time

class RedisLock:
def __init__(self, client, key):
self.client = client
self.key = key

def acquire(self, timeout=10):
identifier = str(time.time())
end = time.time() + timeout
while time.time() < end:
if self.client.set(self.key, identifier, nx=True, ex=30):
return identifier
time.sleep(0.001)
return False

def release(self, identifier):
unlock_script = """
if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end"""
self.client.eval(unlock_script, 1, self.key, identifier)

12.2 Go实现

package main

import (
"context"
"fmt"
"time"
"github.com/go-redis/redis/v8"
)

type RedisLock struct {
client *redis.Client
key string
}

func (l *RedisLock) Acquire(ctx context.Context, expire time.Duration) (bool, error) {
result, err := l.client.SetNX(ctx, l.key, "locked", expire).Result()
if err != nil {
return false, fmt.Errorf("lock acquire failed: %v", err)
}
return result, nil
}

func (l *RedisLock) Release(ctx context.Context) error {
script := redis.NewScript(`
if redis.call("get", KEYS[1]) == "locked" then
return redis.call("del", KEYS[1])
end
return 0
`)
_, err := script.Run(ctx, l.client, []string{l.key}).Result()
return err
}

十三、前沿技术演进

以上方案均经过生产验证，建议根据实际业务场景选择合适的技术组合。对于金融级场景推荐RedLock+自动续期方案，电商秒杀场景建议采用分段锁+本地缓存降级策略。