共计 5589 个字符，预计需要花费 14 分钟才能阅读完成。
自动写代码机器人，免费开通
这篇文章给大家分享的是有关如何使用 Redis 实现一个安全可靠的分布式锁的内容。丸趣 TV 小编觉得挺实用的，因此分享给大家做个参考，一起跟随丸趣 TV 小编过来看看吧。
并发场景下多个进程或线程共享资源的读写，需要保证对资源的访问互斥。在单机系统中，我们可以使用 Java 并发包中的 API、synchronized 关键字等方式来解决；但是在分布式系统下，这些方式不再适用，我们需要自己实现分布式锁。
常见的分布式锁的实现方案有：基于数据库、基于 Redis、基于 Zookeeper 等。作为 Redis 专题的一部分，本文将基于 Redis 聊一聊分布式锁的实现方案。
分析与实现问题分析
分布式锁与 JVM 内置的锁有着共同的目的：让应用程序以预期的顺序访问或操作共享的资源，防止多个线程同时对同一资源操作，导致系统运行紊乱、不可控。常常用于商品库存扣减、优惠券扣减等场景。
理论上来讲，为了保证锁的安全性和有效性，分布式锁至少需要满足以下条件：
互斥性：在同一时间内，仅有一个线程能够获得锁；
无死锁：线程获取锁后，必须保证能够释放，即使线程获取锁后应用程序宕机，也能在限定时间内释放；
加锁和解锁必须是同一个线程；
在实现方式上，分布式锁大体分为三个步骤：
a- 获取资源的操作权；
b- 对资源执行操作；
c- 释放资源的操作权；
无论是 Java 内置的锁，还是分布式锁，也无论使用哪种分布式实现方案，都是围绕 a、c 两个步骤展开。Redis 对于实现分布式锁天然友好，原因如下：
命令处理阶段 Redis 使用单线程处理，同一个 key 同时只有一个线程能够处理，没有多线程竞态问题。
SET key value NX PX milliseconds 命令在不存在 key 的情况下添加具有过期时间的 key，为安全加锁提供支持。
Lua 脚本和 DEL 命令为安全解锁提供可靠支撑。
代码实现
Maven 依赖
dependency
groupId org.springframework.boot /groupId
artifactId spring-boot-starter-data-redis /artifactId
version ${your-spring-boot-version} /version
/dependency配置文件
在 application.properties 增加以下内容，单机版 Redis 实例。
spring.redis.database=0
spring.redis.host=localhost
spring.redis.port=6379RedisConfig
@Configuration
public class RedisConfig {
// 自己定义了一个 RedisTemplate
@Bean
@SuppressWarnings(all)
public RedisTemplate String, Object redisTemplate(RedisConnectionFactory factory)
throws UnknownHostException {
// 我们为了自己开发方便，一般直接使用 String, Object
RedisTemplate String, Object template = new RedisTemplate String,
Object
template.setConnectionFactory(factory);
// Json 序列化配置
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
// String 的序列化
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key 采用 String 的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash 的 key 也采用 String 的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value 序列化方式采用 jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash 的 value 序列化方式采用 jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}RedisLock
@Service
public class RedisLock {
@Resource
private RedisTemplate String, Object redisTemplate;
/**
* 加锁，最多等待 maxWait 毫秒
*
* @param lockKey 锁定 key
* @param lockValue 锁定 value
* @param timeout 锁定时长（毫秒） * @param maxWait 加锁等待时间（毫秒） * @return true- 成功，false- 失败
*/
public boolean tryAcquire(String lockKey, String lockValue, int timeout, long maxWait) { long start = System.currentTimeMillis();
while (true) {
// 尝试加锁
Boolean ret = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, timeout, TimeUnit.MILLISECONDS);
if (!ObjectUtils.isEmpty(ret) ret) {
return true;
}
// 计算已经等待的时间
long now = System.currentTimeMillis();
if (now - start maxWait) {
return false;
}
try { Thread.sleep(200);
} catch (Exception ex) {
return false;
}
}
}
/**
* 释放锁
*
* @param lockKey 锁定 key
* @param lockValue 锁定 value
* @return true- 成功，false- 失败
*/
public boolean releaseLock(String lockKey, String lockValue) {
// lua 脚本
String script = if redis.call(get ,KEYS[1]) == ARGV[1] then return redis.call(del ,KEYS[1]) else return 0 end
DefaultRedisScript Long redisScript = new DefaultRedisScript (script, Long.class);
Long result = redisTemplate.opsForValue().getOperations().execute(redisScript, Collections.singletonList(lockKey), lockValue);
return result != null result 0L;
}
}测试用例
@SpringBootTest
class RedisDistLockDemoApplicationTests {
@Resource
private RedisLock redisLock;
@Test
public void testLock() { redisLock.tryAcquire( abcd , abcd , 5 * 60 * 1000, 5 * 1000);
redisLock.releaseLock( abcd , abcd
}
}安全隐患
可能很多同学（也包括我）在日常工作中都是使用上面的实现方式，看似是稳妥的：
使用 set 命令 NX、PX 选项进行加锁，保证了加锁互斥，避免了死锁；
使用 lua 脚本解锁，防止解除其他线程的锁；
加锁、解锁命令都是原子操作；
其实以上实现的稳妥有个前提条件：单机版 Redis、开启 AOF 持久化方式并设置 appendfsync=always。
但是在哨兵模式和集群模式下可能存在问题，为什么呢？
哨兵模式和集群模式基于主从架构，主从之间通过命令传播实现数据同步，而命令传播是异步的。
所以就存在主节点数据写入成功，在还未通知从节点情况下，主节点就宕机的可能。
当从节点通过故障转移提升为新的主节点后，其他线程就有机会重新加锁成功，导致不满足分布式锁的互斥条件。
官方 RedLock
集群模式下，若集群所有节点稳定运行，不出现故障转移的情况下，安全性是有保障的。但是，没有什么系统能够保证 100% 稳定，基于 Redis 的分布式锁必须考虑容错。
由于主从同步基于异步复制原理，所以哨兵模式和集群模式天生无法满足此条件。为此，Redis 作者专门提出了一种解决方案——RedLock（Redis Distribute Lock）。
设计思路
根据官方文档的说明，把 RedLock 的设计思路进行介绍。
先说环境要求，需要 N（N =3）个独立部署的 Redis 实例，相互之间不需要主从复制、故障转移等技术。
为了获取锁，客户端将按照以下流程进行操作：
获取当前时间（毫秒）作为开始时间 start；
使用相同的 key 和随机 value，按顺序向所有 N 个节点发起获取锁的请求。当向每个实例设置锁时，客户端会使用一个过期时间（小于锁的自动释放时间）。比如锁的自动释放时间是 10 秒，这个超时时间应该是 5 -50 毫秒。这是为了防止客户端在一个已经宕机的实例浪费太多时间：如果 Redis 实例宕机，客户端尽快处理下一个实例。
客户端计算加锁消耗的时间 cost（cost=start-now）。只有客户端在半数以上实例加锁成功，并且整个耗时小于整个有效时间（ttl），才能认为当前客户端加锁成功。
如果客户端加锁成功，那么整个锁的真正有效时间应该是：validTime=ttl-cost。
如果客户端加锁失败（可能是获取锁成功实例数未过半，也可能是耗时超过 ttl），那么客户端应该向所有实例尝试解锁（即使刚刚客户端认为加锁失败）。
RedLock 的设计思路延续了 Redis 内部多种场景的投票方案，通过多个实例分别加锁解决竞态问题，虽然加锁消耗了时间，但是消除了主从机制下的安全问题。
代码实现
官方推荐 Java 实现为 Redisson，它具备可重入特性，按照 RedLock 进行实现，支持独立实例模式、集群模式、主从模式、哨兵模式等；API 比较简单，上手容易。示例如下（直接通过测试用例）：
@Test
public void testRedLock() throws InterruptedException { Config config = new Config();
config.useSingleServer().setAddress( redis://127.0.0.1:6379
final RedissonClient client = Redisson.create(config);
// 获取锁实例
final RLock lock = client.getLock( test-lock
// 加锁
lock.lock(60 * 1000, TimeUnit.MILLISECONDS);
try {
// 假装做些什么事情
Thread.sleep(50 * 1000);
} catch (Exception ex) { ex.printStackTrace();
} finally {
// 解锁
lock.unlock();
}
}Redisson 封装的非常好，我们可以像使用 Java 内置的锁一样去使用，代码简洁的不能再少了。关于 Redisson 源码的分析，网上有很多文章大家可以找找看。
全文总结
分布式锁是我们研发过程中常用的的一种解决并发问题的方式，Redis 是只是一种实现方式。
关键的是要弄清楚加锁、解锁背后的原理，以及实现分布式锁需要解决的核心问题，同时考虑我们所采用的中间件有什么特性可以支撑。了解这些后，实现起来就不是什么问题了。
感谢各位的阅读！关于“如何使用 Redis 实现一个安全可靠的分布式锁”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！
向 AI 问一下细节
