Redis使用Lua脚本和Redisson来保证库存扣减中的原子性和一致性

前言

背景：最近有社群技术交流的同学，说面试被问到商品库存扣减的问题。我大概整理了一下内容，方便大家理解。其实无外乎就是分布式锁和Redis命令的原子性问题。

在分布式系统中，保证数据的原子性和一致性是一个关键问题。特别是在库存扣减等场景中，确保操作的原子性是至关重要的，以避免数据不一致和并发冲突的问题。为了解决这个挑战，我们可以利用 Redis 数据库的强大功能来实现库存扣减的原子性和一致性。

本博客将介绍两个关键技术：Redis Lua脚本和Redisson，它们在库存扣减场景中的应用。Lua脚本是一种嵌入在 Redis 服务器中执行的脚本语言，具有原子性执行和高效性能的特点。而Redisson是一个基于 Redis 的分布式 Java 对象和服务框架，提供了丰富的功能和优势。

所以无论是对于中小型企业还是大型互联网公司，保证库存扣减的原子性和一致性都是至关重要的。本博客将帮助读者全面了解如何利用 Redis Lua脚本和 Redisson 来实现这一目标，为他们的分布式系统提供可靠的解决方案。让我们一起深入研究这些强大的工具，提升我们的分布式系统的性能和可靠性。

1.使用SpringBoot +Redis 原生实现方式

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;@Component
public class StockService {@Resourceprivate RedisTemplate<String, Object> redisTemplate;// 扣减商品库存public void decreaseStock(String productId, int quantity) {String lockKey = "lock:" + productId;String stockKey = "stock:" + productId;ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();Boolean acquiredLock = valueOperations.setIfAbsent(lockKey, "locked");try {if (acquiredLock != null && acquiredLock) {// 获取锁成功，设置锁的过期时间，防止死锁redisTemplate.expire(lockKey, 5, TimeUnit.SECONDS);Integer currentStock = (Integer) valueOperations.get(stockKey);if (currentStock != null && currentStock >= quantity) {int newStock = currentStock - quantity;valueOperations.set(stockKey, newStock);System.out.println("库存扣减成功");} else {System.out.println("库存不足，无法扣减");}} else {System.out.println("获取锁失败，其他线程正在操作");}} finally {// 释放锁if (acquiredLock != null && acquiredLock) {redisTemplate.delete(lockKey);}}}
}  

我们思考一下，以上这种写法存在几个问题，这种问题

1. 锁的释放问题：在当前代码中，锁的释放是通过判断获取锁成功与否来决定是否释放锁。然而，如果在执行redisTemplate.expire设置锁的过期时间之后，代码发生异常导致没有执行到锁的释放部分，将会导致锁无法及时释放，进而可能导致其他线程无法获取锁。为了解决这个问题，可以考虑使用Lua脚本来实现原子性的获取锁和设置过期时间。

2. 锁的重入问题：当前代码中，没有对锁的重入进行处理。如果同一个线程多次调用decreaseStock方法，会导致获取锁失败，因为锁已经被当前线程占用。为了解决这个问题，可以考虑使用ThreadLocal或者维护一个计数器来记录锁的重入次数，以便在释放锁时进行正确的处理。

解决方法
对于上述代码的优化，可以考虑以下几点：

1. 使用setIfAbsent方法设置锁，并将锁的过期时间与设置锁合并为一个原子操作，以避免在获取锁后再次操作Redis的时间开销。可以使用opsForValue().setIfAbsent(lockKey, "locked", 5, TimeUnit.SECONDS)来实现。这样可以确保获取锁和设置过期时间是一个原子操作，避免了两次Redis操作的时间间隔。

2. 使用lua脚本来实现锁的释放，以确保释放锁的原子性。通过使用execute方法执行lua脚本，可以将锁的释放操作合并为一个原子操作。以下是示例代码：

String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(luaScript, Long.class);
Long releasedLock = redisTemplate.execute(redisScript, Collections.singletonList(lockKey), "locked");
if (releasedLock != null && releasedLock == 1) {// 锁释放成功
}

3. 使用Redisson等可靠的分布式锁框架，它们提供了更丰富的功能和可靠性，并且已经解决了很多与分布式锁相关的问题。这些框架可以简化代码并提供更强大的锁管理功能，例如重入锁、公平锁、红锁等。你可以在项目中引入Redisson等框架，并使用它们提供的分布式锁功能。

2.使用redisson方式实现

可能有一些同学对Redisson不太了解，我大概讲解一下他的一些优秀之处。
Redisson 是一个基于 Redis 的分布式 Java 对象和服务框架，它提供了丰富的功能和优势，使得在分布式环境中使用 Redis 更加方便和可靠。可以这么说，Redisson 是目前最牛逼最强的基于Redis的分布式锁工具，没有之一，所以大家可以在项目中放心大胆的使用，有问题再说问题，不要太过羁绊。

Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

1. 分布式锁： 提供了可重入锁、公平锁、联锁、红锁等多种分布式锁的实现，可以用于解决并发控制问题。它支持锁的自动续期和异步释放，可以防止锁的过期导致的问题，并提供了更高级的功能如等待锁、超时锁等。

2. 分布式集合： 提供了一系列分布式集合的实现，如分布式列表、集合、有序集合、队列、阻塞队列等。这些分布式集合可以安全地在多个节点之间共享和操作数据，提供了高效的数据存储和访问机制。

3. 分布式对象：Redisson 支持在分布式环境中操作 Java 对象。它提供了分布式映射、分布式原子变量、分布式计数器等功能，可以方便地对分布式对象进行存储、操作和同步。

4. 优化的 Redis 命令：Redisson 通过优化 Redis 命令的调用方式，提供了更高效的数据访问。它使用了线程池和异步操作，可以在一次网络往返中执行多个 Redis
   命令，减少了网络延迟和连接数，提高了性能和吞吐量。

5. 可扩展性和高可用性：Redisson 支持 Redis 集群和哨兵模式，可以轻松应对大规模和高可用性的需求。它提供了自动的故障转移和主从切换机制，确保在节点故障时系统的可用性和数据的一致性。
   一个基于Redis实现的分布式工具，有基本分布式对象和高级又抽象的分布式服务，为每个试图再造分布式轮子的程序员带来了大部分分布式问题的解决办法。

吹了那么多概念，请show me code。ok 接下来我们使用Redisson库实现的库存的原子性和一致性。

1. 添加依赖：在pom.xml文件中添加以下依赖以使用Redisson库来实现分布式锁。

<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.16.1</version>
</dependency>

2. 配置Redisson：在Spring Boot的配置文件中添加Redisson的配置，例如application.properties。

# Redisson配置
spring.redisson.config=classpath:redisson.yaml

在resources目录下创建redisson.yaml文件，并配置Redis连接信息和分布式锁的相关配置。以下是一个示例配置：

singleServerConfig:address: "redis://localhost:6379"password: nulldatabase: 0connectionPoolSize: 64connectionMinimumIdleSize: 10subscriptionConnectionPoolSize: 50dnsMonitoringInterval: 5000lockWatchdogTimeout: 10000

3. 创建一个名为StockService的服务类，修改decreaseStock方法来使用分布式锁：

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;import java.util.concurrent.TimeUnit;@Service
public class StockService {private static final String STOCK_KEY = "stock:product123";private static final String LOCK_KEY = "lock:product123";@Autowiredprivate ReactiveRedisTemplate<String, String> redisTemplate;@Autowiredprivate RedissonClient redissonClient;public Mono<Boolean> decreaseStock(int quantity) {RLock lock = redissonClient.getLock(LOCK_KEY);return Mono.fromCallable(() -> {//==核心代码start==try {boolean acquired = lock.tryLock(1, 10, TimeUnit.SECONDS);if (acquired) {Long stock = redisTemplate.opsForValue().get(STOCK_KEY).block();if (stock != null && stock >= quantity) {redisTemplate.opsForValue().decrement(STOCK_KEY, quantity).block();return true;}}return false;} finally {lock.unlock();}//==核心代码结束==});}public Mono<Long> getStock() {return redisTemplate.opsForValue().get(STOCK_KEY).map(stock -> stock != null ? Long.parseLong(stock) : 0L);}
}

在StockService中，我们首先通过redissonClient.getLock方法获取一个分布式锁对象 RLock，并使用tryLock方法尝试获取锁。如果成功获取到锁，则执行库存扣减操作。在操作完成后，释放锁。通过使用分布式锁，我们确保了在并发场景下只有一个线程可以执行库存扣减操作，从而保证了原子性和一致性。

3. 使用Redis+Lua脚本实现

使用Lua脚本来实现库存扣减的原子性操作 。使用了Spring Data Redis提供的RedisTemplate来与Redis进行交互，并使用DefaultRedisScript定义了Lua脚本。通过ScriptExecutor执行Lua脚本，将库存扣减的逻辑放在脚本中实现。

decreaseStock方法中，我们 定义了Lua脚本，然后创建了一个DefaultRedisScript对象，并指定脚本返回值的类型为Boolean。接下来，我们通过scriptExecutor.execute方法执行Lua脚本，并传递脚本、键（STOCK_KEY）和参数（quantity）作为参数。

getStock方法则使用Mono.fromSupplier来获取当前库存数量，与Lua脚本无关。

3.1 lua脚本

代码逻辑

1. 通过redis.call('GET', KEYS[1])从Redis中获取键（KEYS[1]）对应的库存数量，并使用tonumber将其转换为数字类型。
2. 检查库存是否足够进行扣减，即判断stock是否存在且大于等于传入的扣减数量（ARGV[1]）。
3. 如果库存足够，使用redis.call('DECRBY', KEYS[1], ARGV[1])扣减库存。
4. 返回true表示扣减成功，否则返回false表示扣减失败。

-- 从Redis中获取当前库存
local stock = tonumber(redis.call('GET', KEYS[1]))-- 检查库存是否足够扣减
if stock and stock >= tonumber(ARGV[1]) then-- 扣减库存redis.call('DECRBY', KEYS[1], ARGV[1])return true -- 返回扣减成功
elsereturn false -- 返回扣减失败
end

3.2 与SpringBoot集成

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.core.script.ScriptExecutor;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;import java.util.Collections;@Service
public class StockService {private static final String STOCK_KEY = "stock:product123";@Autowiredprivate RedisTemplate<String, String> redisTemplate;@Autowiredprivate ScriptExecutor<String> scriptExecutor;public Mono<Boolean> decreaseStock(int quantity) {String script = "local stock = tonumber(redis.call('GET', KEYS[1]))\n" +"if stock and stock >= tonumber(ARGV[1]) then\n" +"    redis.call('DECRBY', KEYS[1], ARGV[1])\n" +"    return true\n" +"else\n" +"    return false\n" +"end";RedisScript<Boolean> redisScript = new DefaultRedisScript<>(script, Boolean.class);return scriptExecutor.execute(redisScript, Collections.singletonList(STOCK_KEY), String.valueOf(quantity));}public Mono<Long> getStock() {return Mono.fromSupplier(() -> {String stock = redisTemplate.opsForValue().get(STOCK_KEY);return stock != null ? Long.parseLong(stock) : 0L;});}
}

4. Lua脚本方式和Redisson的方式对比

在使用Lua脚本执行库存扣减操作时，通常不需要显式地加锁。这是因为Redis执行Lua脚本的机制保证了脚本的原子性。

当Redis执行Lua脚本时，会将整个脚本作为一个单独的命令进行执行。在执行期间，不会中断脚本的执行，也不会被其他客户端的请求打断。这使得Lua脚本在执行期间是原子的，即使在高并发的情况下也能保证操作的一致性。

因此，在上述的Lua脚本中，我们没有显式地加锁来保护库存扣减操作。通过使用Lua脚本，我们充分利用了Redis的原子性操作特性，避免了显式加锁的开销和复杂性。

需要注意的是，如果有其他并发操作也需要对库存进行扣减或修改，可能需要考虑加锁机制来保证操作的原子性。这种情况下，可以使用分布式锁来控制对库存的访问，以确保并发操作的正确性。

使用Redisson的方式和使用Lua脚本的方式在实现库存扣减时有一些不同之处。 我们做成一个表格可以清晰的对比一下，方便理解记忆，其实在项目的真正实践过程中，这两种方式也是比较常见的。但是具体使用哪一种要看大家公司的技术积累和使用偏好。
所以我们总结一下。选择适当的方式取决于具体的需求和场景。如果你需要更灵活的控制、更多的分布式功能或者对性能要求较高，那么使用Redisson库可能是一个不错的选择。而如果你希望简化实现并减少依赖，而且对性能要求不是非常高，那么使用Lua脚本可能更为合适。

方式	实现复杂性	灵活性	性能开销	分布式环境功能
Redisson库	需要额外的依赖和配置，编写相关代码	提供更多功能选项，如超时设置、自动续期等	可能涉及网络通信和分布式锁管理的性能开销	提供更丰富的分布式功能
Lua脚本	无额外依赖，只需编写Lua脚本	相对简单，专注于库存扣减逻辑	通常具有较低延迟和较高性能	专注于库存扣减操作，无其他分布式功能的支持

5. 源码地址

https://github.com/wangshuai67/Redis-Tutorial-2023

6. Redis从入门到精通系列文章

• 《【Redis实践篇】使用Redisson 优雅实现项目实践过程中的5种场景》
• 《Redis使用Lua脚本和Redisson来保证库存扣减中的原子性和一致性》
• 《SpringBoot Redis 使用Lettuce和Jedis配置哨兵模式》
• 《Redis【应用篇】之RedisTemplate基本操作》
• 《Redis 从入门到精通【实践篇】之SpringBoot配置Redis多数据源》
• 《Redis 从入门到精通【进阶篇】之三分钟了解Redis HyperLogLog 数据结构》
• 《Redis 从入门到精通【进阶篇】之三分钟了解Redis地理位置数据结构GeoHash》
• 《Redis 从入门到精通【进阶篇】之高可用哨兵机制(Redis Sentinel)详解》
• 《Redis 从入门到精通【进阶篇】之redis主从复制详解》
• 《Redis 从入门到精通【进阶篇】之Redis事务详解》
• 《Redis从入门到精通【进阶篇】之对象机制详解》
• 《Redis从入门到精通【进阶篇】之消息传递发布订阅模式详解》
• 《Redis从入门到精通【进阶篇】之持久化 AOF详解》
• 《Redis从入门到精通【进阶篇】之持久化RDB详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构字典(Dictionary)详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构快表QuickList详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构简单动态字符串(SDS)详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构压缩列表(ZipList)详解》
• 《Redis从入门到精通【进阶篇】之数据类型Stream详解和使用示例》
  大家好，我是冰点，今天的Redis【实践篇】之Redis使用Lua脚本和Redisson来保证库存扣减中的原子性和一致性，全部内容就是这些。如果你有疑问或见解可以在评论区留言。

7. 参考文档

redisson 参考文档 https://redisson.org/documentation.html