Redis下篇--分布式锁
Redis下篇--分布式锁本文示例代码见GITEE仓库中的【redis-analysis】
地址：https://gitee.com/quercus-sp204/new-technology/tree/master/redis-analysis
1.基本介绍
Redis分布式锁是一种基于Redis实现的跨进程互斥机制，用于在分布式系统中控制多个服务/节点对共享资源的并发访问，确保同一时刻只有一个客户端能执行关键操作（如修改共享数据、执行任务等）。
在单机系统中，我们可以用线程锁（如Java的synchronized或ReentrantLock）保证并发安全。但在分布式系统中【项目集群部署】，服务部署在多台机器上，跨进程的共享资源无法通过本地锁保护。如下图1所示：
此时就需要分布式锁。我们需要用它来：
[*]解决分布式竞争条件：当多个服务实例同时操作共享资源（如库存扣减、订单状态更新）时，可能因并发写入导致数据错误（例如超卖）。分布式锁确保同一时刻只有一个实例能执行操作。
[*]替代低效方案：传统方案（如数据库行锁）在并发高时性能差，而Redis作为内存数据库，高性能（10万+ QPS） 和原子操作特性使其成为理想的分布式锁实现基础。
[*]保证操作原子性：分布式锁将临界区操作（如“查询+修改”）封装为原子操作，避免多个客户端交叉执行引发的逻辑错误。
大致示意图如下图2所示：（就是多加了一层！）
上面说了，用于分布式系统的，我在这里抛出一个小问题：现在也有很多应用不是微服务架构的，我们是不是没有必要上分布式锁呢？熟悉java的人都知道，我们可以用java自带的synchronized、ReentrantLock锁呀，我们为什么要这个分布式锁呢？好像确实有些许道理。这个问题先放一放。
介绍完了其基本信息，下面给出其在实际生产中的应用场景：
[*]防止重复操作：用户重复提交订单：方案：用订单ID+操作类型作为锁key，确保同一订单的支付操作只执行一次。
[*]高并发库存扣减：
# 伪代码示例【大致流程】
lock_key = "stock_lock:product_123"
if redis.set(lock_key, "1", nx=True, ex=5):# 获取锁
try:
stock = db.query("SELECT stock FROM products WHERE id=123")
if stock > 0:
db.execute("UPDATE products SET stock = stock - 1 WHERE id=123")
finally:
redis.del(lock_key)# 释放锁
[*]分布式任务调度：多个节点竞争执行定时任务（如每天0点统计报表，或者是每天定时数据库做一些数据修改之类的），在要执行定时任务的时候，我们可以用任务名作为key，抢到锁的节点执行任务。
等等场景都可以用到分布式锁。其实分布式锁实现方式不只有Redis可以实现，还可以用其他方案来实现，比如说
我们可以用数据库来实现分布式锁，通过SELECT ... FOR UPDATE对数据库记录加行锁，亦或者是在表中增加版本号字段，更新时校验版本号（CAS机制）；但是频繁加锁导致数据库IO压力大，高并发下延迟显著，同时事务未提交或超时可能引发死锁。
还可以使用Zookeeper来创建临时结点来达到类似的效果，创建临时顺序节点，最小序号节点获锁，其他节点监听前序节点删除事件，节点宕机时临时节点自动删除，避免死锁。但是相比于redis来说，节点创建/删除及事件通知的开销较大（低于Redis），同时可能需维护ZooKeeper集群，开发成本较高。
实际生产要将成本和系统复杂度综合起来考虑，最后再决定采用哪种方案，但是Redis这个中间件对于现在的很多系统来说，我认为是用得非常普遍了吧。在综合考虑性能、系统复杂度、以及业界普遍方案，本文就以Redis实现分布式锁展开了。
2. Redis实现分布式锁
2.1 基础版setnx
SETNX lock_key 1# 尝试获取锁
DEL lock_key # 释放锁
@Component
public class BaseLock {
@Resource
private RedisTemplate<String, Object> redisTemplate;
public boolean tryLock(String lockName) {
Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockName, "lock");
return lock != null && lock;
}
public void unLock(String lockName) {
redisTemplate.delete(lockName);
}
}
// 测试使用
public void test() {
// 1.加锁
try {
bool res = baseLock.tryLock(key);
if ( res ) {
do......
} else {
获取锁失败
}
} catch( e ) {
........
} finally{
baseLock.unLock(key) // 如果这一行死活执行不了
}
}这绝对是有一点问题的，如果有异常原因，在执行baseLock.unLock(key)这行代码的时候出现异常怎么办，可能这个时候网络异常？Redis客户端断开了？等等原因，导致该key没有被删掉，那么就会有死锁问题。下面为了避免这个问题，我们把自动删除key的操作，加一个过期时间，让redis-server为我们兜一下底。
2.2 过期时间
在设置过期时间这一点上，需要提前说明一下，Redis是有EXPIRE这个命令的，它为锁设置过期时间，避免死锁（如 EXPIRE lock_key 10），如果是先setnx，再expire，这就是两条命令了，所以我们需要保证两条命令的原子性。但是Redis如今都发展到版本7.x了，Redis早就提供了原子命令：
SET lock_key unique_value NX EX 10 #同时完成锁设置和过期时间，避免非原子性问题本文就直接用上面这个了。
// 方案二：set + nx + ex
public boolean tryLock2(String lockName, int t) {
Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockName, "lock", Duration.of(t, ChronoUnit.SECONDS));
return lock != null && lock;
}
public void unLock2(String lockName) {
redisTemplate.delete(lockName);
}
// 测试使用
public void test() {
// 1.加锁
try {
// 指定key的有效时间 假设time=5
bool res = baseLock.tryLock2(key, time);
if ( res ) {
do...... // 【但是这里执行6秒喔】
} else {
获取锁失败
}
} catch( e ) {
........
} finally{
baseLock.unLock2(key) //【释放锁处】
}
}上面这个方案，就算是【释放锁处】死活执行不了，有过期时间为我们兜底，不用担心死锁问题，过期时间到了就行了。上面仍然会有问题，且听细细道来。
看上面测试代码的注释 do...... // 【但是这里执行6秒喔】，就是说业务执行时间超过了锁的过期时间，就会导致如下问题：
请按照顺序看上面的流程，我们可以知道发生了这样一件事情：那就是线程1把线程2的锁给删了，出现误删的现象，然后后面就可能会引发一连串的错误了。
下面来看方案3，前面两个方案的value我们都没有用到，现在要用到了：
// 方案三：set + uuid
public boolean tryLock3(String lockName, String value, int t) {
Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockName, value, Duration.of(t, ChronoUnit.SECONDS));
return lock != null && lock;
}
public void unLock3(String lockName, String value) {
String lock = (String) redisTemplate.opsForValue().get(lockName);
if ( lock != null && lock.equals(value)) { // 比较一下value，是不是自己的
redisTemplate.delete(lockName);
}
}
public void test() {
String lockName = "lock";
String uuid = UUID.randomUUID().toString();
try {
boolean b = tryLock3(lockName, uuid, 5);
if (b) {
// 获取锁成功
// 逻辑代码
Thread.sleep(5000);
} else {
// 获取锁失败
}
} catch (Exception e ) {
// ...
} finally {
unLock3(lockName, uuid);
}
}虽然这样看起来不会释放别人的锁了，但是如果业务超时，还是会出现有多个线程进入临界区的情况，这是不希望看到的。还有一个小问题，就是unLock3的那两步不是原子操作，极端情况也会出现误删问题：下面给出deepseek的极端情况推演图
2.3 lua脚本
所以要保证原子性。我们使用redis lua脚本来保证这两部操作的原子性：
private static final String UNLOCK_SCRIPT =
"if redis.call('get', KEYS) == ARGV then " +
" return redis.call('del', KEYS) " +
"else " +
" return 0 " +
"end";
// 删除锁的时候用lua脚本 unLock3优化版
public void unLock3(String lockName, String value) {
DefaultRedisScript<Long> script = new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class);
redisTemplate.execute(script, Collections.singletonList(lockName), value);
}这里顺带提一嘴，Redis不是还有事务吗？我们可以用事务吗：
public void unsafeUnlockWithTransaction(String lockName, String value) {
redisTemplate.execute(new SessionCallback<>() {
@Override
public Object execute(RedisOperations operations) {
operations.watch(lockName);
String lockValue = (String) operations.opsForValue().get(lockName);
if (value.equals(lockValue)) {
operations.multi();
operations.delete(lockName);
operations.exec(); // 事务提交
} else {
operations.unwatch();
}
return null;
}
});
}那么，这个Redis事务在第三章补充一下，在此处先放一下。再次回到2.2小节的问题，那就是业务时间 > 锁持有时间，还是会导致多个线程进入临界区的情况，所以这个过期时间是我们要着重考虑的问题了。
2.4 锁续期
对于这个问题，我们可以有这样的实现思路，启动一个线程在后台对key进行续期不就好了？可以我们可以利用Java的周期任务线程来完成这个效果，简单示例代码如下：
@Component
@Slf4j
public class AutoExtensionLock {
@Resource
private RedisTemplate<String, Object> redisTemplate;
// 锁的队列
private static ConcurrentLinkedQueue<RedisLockHolder> keys = new ConcurrentLinkedQueue<>();
private static final ScheduledExecutorService SCHEDULER =
new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().daemon(true).namingPattern("redis-AutoExtensionLock-%d").build());
@PostConstruct
public void init() {
// 定时检查锁的到期时间
SCHEDULER.scheduleAtFixedRate(() -> {
Iterator<RedisLockHolder> iterator = keys.iterator();
while (iterator.hasNext()) {
RedisLockHolder redisLockHolder = iterator.next();
log.info("redis-AutoExtensionLock-info, {}", redisLockHolder.toString());
try { //try-catch起来否者报错后定时任务将不会再运行
if (redisLockHolder.getEndTime() >= System.currentTimeMillis()) {
Object v = redisTemplate.opsForValue().get(redisLockHolder.getKey());
if ( v == null ) { // 锁不存在
iterator.remove();
} else {
long holderExpireTime = redisLockHolder.getEndTime();
long now = System.currentTimeMillis();
long during = redisLockHolder.getDuring();
int maxRetry = redisLockHolder.getMaxRetry();
int nowCount = redisLockHolder.getNowCount();
long l = holderExpireTime - now; // 锁的剩余时间
if ( nowCount == maxRetry ) {
log.info("redis-AutoExtensionLock-error, {}", "重试次数已满");
iterator.remove();
} else if ( l <= 0 ) {
log.info("redis-AutoExtensionLock-error, {}", "锁已过期");
iterator.remove();
}
// 如果 l <= during / 3
else if ( l <= during / 3.0 ) {
log.info("redis-AutoExtensionLock-info, {}", "锁即将过期，开始自动续期");
redisTemplate.expire(redisLockHolder.getKey(), redisLockHolder.getDuring(), TimeUnit.SECONDS);
redisLockHolder.setNowCount(nowCount + 1);
redisLockHolder.setEndTime(now + during);
}
}
}
else {
iterator.remove();
}
} catch (Exception e) {
// ....
log.info("redis-AutoExtensionLock-error, {}", e.toString());
iterator.remove();
}
}
}, 0, 3000, TimeUnit.MILLISECONDS);
}
// 尝试获取锁
public boolean tryLock4(String lockName, String value, long t ) {
Boolean b = redisTemplate.opsForValue().setIfAbsent(lockName, value, Duration.of(t, ChronoUnit.SECONDS));
if (b != null && b) {
keys.add(new RedisLockHolder(lockName, value, t * 1000,System.currentTimeMillis() + (t * 1000), 0, 5));
return true;
}
return false;
}
// 释放锁
private static final String UNLOCK_SCRIPT =
"if redis.call('get', KEYS) == ARGV then " +
" return redis.call('del', KEYS) " +
"else " +
" return 0 " +
"end";
public void unlock4(String lockName, String value) {
DefaultRedisScript<Long> script = new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class);
redisTemplate.execute(script, Collections.singletonList(lockName), value);
}
@Data
@AllArgsConstructor
@NoArgsConstructor
private static class RedisLockHolder {
public static final int MAX_RETRY = 5;
private String key;
private String value;
private long during; // 锁的持续时间 --- 毫秒
private long endTime; // 到期时间
private int nowCount; // 当前重试次数
private int maxRetry; // 重试次数
}
}3. 补充点
3.1 redis事务
上文说到了Redis的事务问题，本章探讨一下Redis的事务：
事务本质是一组命令的集合，支持一次执行多个命令，一个事务中所有命令都会被序列化。在redis事务的执行过程，会按照顺序串行化执行队列中的命令，其他客户端提交的命令请求不会插入到事务执行命令序列中。
redis的事务指令有3个关键字，分别是：
[*]multi:开启事务
[*]exec:执行事务
[*]discard:取消事务
[*]watch：监视Key改变，用于实现乐观锁。如果监视的Key的值改变，事务最终会执行失败。在事务开启前使用
[*]unwatch：放弃监视。
通过multi，当前客户端就会开启事务，后续用户键入的都指令都会保证到队列中暂不执行，当用户键入exec后，这些指令都会按顺序执行。 需要注意的是，若开启multi后输入若干指令，客户端输入discard，则之前的指令通通取消执行。
总结说：Redis事务就是一次性、顺序性、排他性的执行一个队列中的一系列命令
RLock lock = redisson.getLock("orderLock");
lock.lock(); // 获取锁（自动续期）
try {
// 业务操作
} finally {
lock.unlock(); // 原子化释放
}那么Redis事务满足我们熟悉的事务四大特性吗？
对于隔离性来说，Redis是单线程执行命令的，并且执行事务时是对事务队列中的命令依次执行，因此Redis不会出现隔离性问题。
持久性那就不必多说了。
重点看一下原子性：
<dependency>
<groupId>org.redisson</groupId>
redisson-spring-boot-starter</artifactId>
<version>3.45.1</version>
</dependency>当命令输入错误会在执行时直接报错，这种情况下能够满足原子性，
当运行时出现错误时，会执行到具体命令时才报错，这种情况下除了报错的命令不执行，事务中其他正常的命令会执行，不能满足原子性
为什么这么做？
[*]使用Redis命令语法错误，或是将命令运用在错误的数据类型键上(如对字符串进行加减乘除等)，从而导致业务数据有问题，这种情况认为是编程导致的错误，应该在开发过程中解决，避免在生产环境中发生；
[*]由于不用支持回滚功能，Redis内部简单化，而且还比较快；
多数事务失败是由语法错误或者数据结构类型错误导致的，语法错误说明在命令入队前就进行检测的，而类型错误是在执行时检测的，Redis为提升性能而采用这种简单的事务，这是不同于关系型数据库的，特别要注意区分。Redis之所以保持这样简易的事务，完全是为了保证高并发下的核心问题——性能。
接下来看一下watch命令：【最前面的数字，表示命令执行顺序】
客户端一：
public boolean tryLock(String lockName) {
return redissonClient.getLock(lockName).tryLock();
}
public void unLock(String lockName) {
redissonClient.getLock(lockName).unlock();
}
// 带超时的锁
public boolean tryLock(String lockName, long waitTime, long leaseTime) {
try {
return redissonClient.getLock(lockName).tryLock(waitTime, leaseTime, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.error("tryLock 出现异常", e);
return false;
}
}客户端二：
// 首先是getLock
// 【Redisson.java】
public final class Redisson implements RedissonClient {
@Override
public RLock getLock(String name) {
// 返回的是RLock类型【具体类型是RedissonLock】
return new RedissonLock(commandExecutor, name);
}
}
// 然后就是tryLock方法了
// 【RedissonLock.java】
public class RedissonLock extends RedissonBaseLock {
@Override
public boolean tryLock() {
// 1.tryLockAsync()
// 2.get( RFuture<Boolean> )
// get( xxx ) 阻塞到tryLockAsync()返回
return get(tryLockAsync());
}
}
// 【RedissonBaseLock.java】
public abstract class RedissonBaseLock extends RedissonExpirable implements RLock {
@Override
public RFuture<Boolean> tryLockAsync() {
// 这里调用的tryLockAsync()方法，具体实现在【RedissonLock.java】
// 这个方法是RLockAsync接口里面定义的
// RLock接口继承了RLockAsync接口，所以这里可以看到这个方法
return tryLockAsync(Thread.currentThread().getId());
}
}
// 现在又回到 【RedissonLock.java】
@Override
public RFuture<Boolean> tryLockAsync(long threadId) {
// 先 () -> tryAcquireOnceAsync(-1, -1, null, threadId)
// 返回RFuture<Boolean>是execute方法返回的
return getServiceManager().execute(() -> tryAcquireOnceAsync(-1, -1, null, threadId));
}
private <T> void execute(AtomicInteger attempts, CompletableFuture<T> result, Supplier<CompletionStage<T>> supplier) {
// 这里一行执行的是传进来的supplier.....也就是() -> tryAcquireOnceAsync(-1, -1, null, threadId)
CompletionStage<T> future = supplier.get();
future.whenComplete((r, e) -> {
if (e != null) { // 有异常
if (.....) {
....
// 重试
newTimeout(t -> execute(attempts, result, supplier),
config.getRetryInterval(), TimeUnit.MILLISECONDS);
return;
}
result.completeExceptionally(e);
return;
}
// 没有异常，ok,future设置为完成
result.complete(r);
});
}
// () -> tryAcquireOnceAsync(-1, -1, null, threadId)
private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
CompletionStage<Boolean> acquiredFuture;
// 如果给的所持有时间 > 0
if (leaseTime > 0) {
//场景：用户显式指定锁超时时间（如 lock.lock(10, TimeUnit.SECONDS)）
acquiredFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
} else {
// 如果 <= 0, 由于我们最外层是调用的tryLock(),没有带任何参数,走到这里的话，leaseTime传过来的是-1
// 在该方法下面看tryLockInnerAsync方法
acquiredFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
}
acquiredFuture = handleNoSync(threadId, acquiredFuture);
CompletionStage<Boolean> f = acquiredFuture.thenApply(acquired -> {
// lock acquired
if (acquired) {
if (leaseTime > 0) { // 如果传进来的锁持有时间>0
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
// 如果传进来的锁持有时间 <= 0
// 会启动看门狗喔
scheduleExpirationRenewal(threadId);
}
}
return acquired;
});
return new CompletableFutureWrapper<>(f);
}
// 可以看到底层是用了lua脚本喔
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
return evalWriteSyncedNoRetryAsync(getRawName(), LongCodec.INSTANCE, command,
"if ((redis.call('exists', KEYS) == 0) " + // 情况1：锁不存在
"or (redis.call('hexists', KEYS, ARGV) == 1)) then " + // 情况2：锁已被当前线程持有
"redis.call('hincrby', KEYS, ARGV, 1); " + // 重入计数+1
"redis.call('pexpire', KEYS, ARGV); " + // 设置/刷新过期时间
"return nil; " + // 返回nil表示成功
"end; " +
"return redis.call('pttl', KEYS);", // 返回锁剩余时间（毫秒）
Collections.singletonList(getRawName()), // KEYS = 锁名称
unit.toMillis(leaseTime),// ARGV = 锁持有时间（毫秒）
getLockName(threadId));// ARGV = 线程标识符（UUID:threadId）
}上面通过watch监视指定Redis Key ( test1 )，如果在事务执行之前没有改变，就执行成功，如果发现对应值发生改变，事务就会执行失败
看到这里，在2.3节可以用事务吗？我认为在逻辑不出错的情况下，是可以用的，需要我们编程人员编码的时候，来确认每一条命令都确保逻辑上都是正确的。
但是我们能用Lua脚本解决的原子性问题，优先用Lua
3.2 tryLock 和 lock
上面源码分析的都是tryLock的，其实还有lock方法，那么这二者有什么区别呢？
（1）返回值： lock() 是没有返回值的；tryLock() 的返回值是 boolean。
（2）时机：lock() 一直等锁释放；tryLock() 获取到锁返回true，获取不到锁并直接返回false。
（3）tryLock() 是可以被打断的，被中断的；lock是不可以。
key: test
value:
----field: 285475da-9152-4c83-822a-67ee2f116a79:52 [线程ID]
----val: 1 [重入次数]实际二者使用
// 接上文
// 会启动看门狗喔
// 【RedissonLock.java】
scheduleExpirationRenewal(threadId);
// 这个方法其实是 【RedissonBaseLock.java】里面的
protected void scheduleExpirationRenewal(long threadId) {
renewalScheduler.renewLock(getRawName(), threadId, getLockName(threadId));
}
// 【LockRenewalScheduler.java】
public void renewLock(String name, Long threadId, String lockName) {
reference.compareAndSet(null, new LockTask(internalLockLeaseTime, executor, batchSize));
LockTask task = reference.get();
task.add(name, lockName, threadId);
}
// ...
// 到这里
// RenewalTask.java
final void add(String rawName, String lockName, long threadId, LockEntry entry) {
addSlotName(rawName);
LockEntry oldEntry = name2entry.putIfAbsent(rawName, entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId, lockName);
} else {
if (tryRun()) {
schedule(); // 这里
}
}
}
public void schedule() {
if (!running.get()) {
return;
}
long internalLockLeaseTime = executor.getServiceManager().getCfg().getLockWatchdogTimeout();
//定时任务 是 lockWatchdogTimeout 的1/3时间去执行 renewExpirationAsync【默认就是10秒嘛】
executor.getServiceManager().newTimeout(this, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
}
// ServiceManager.java
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
try {
return timer.newTimeout(task, delay, unit);
} catch (IllegalStateException e) {
....
}
}
// 然后我们发现，来到netty包下面了 io.netty.util
// public class HashedWheelTimer implements Timer {....}
@Override
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
// delay参数是internalLockLeaseTime / 3
}end. 参考
[*]https://mp.weixin.qq.com/s/nrCO8GZBJrLQis98bMaRhg
[*]https://mp.weixin.qq.com/s/UzMTAqVy5MXxmV9rXdgyPg
[*]https://www.cnblogs.com/jackson0714/p/redisson.html
[*]https://mp.weixin.qq.com/s/qVPT-e-gOXsqQ3pMVIEK7A 【Redis事务】
[*]https://segmentfault.com/a/1190000044686369 【思否- redis事务】
[*]https://blog.csdn.net/jiayi_yao/article/details/124689937 【Redis事务】
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]