Redis从入门到精通(九)Redis实战(六)基于Redis队列实现异步秒杀下单

↑↑↑请在文章开头处下载测试项目源代码↑↑↑

前言

Redis实战系列文章：

Redis从入门到精通(四)Redis实战(一)短信登录
Redis从入门到精通(五)Redis实战(二)商户查询缓存
Redis从入门到精通(六)Redis实战(三)优惠券秒杀
Redis从入门到精通(七)Redis实战(四)库存超卖、一人一单与Redis分布式锁
Redis从入门到精通(八)Redis实战(五)分布式锁误删与原子性问题、Redisson

4.5 分布式锁-Redisson

上一节对Redisson进行了快速入门，并分析了可重入锁的基本原理，下面继续研究一些Redisson的几个功能。

4.5.4 Redission锁重试

// org.redisson.RedissonLock#lock()long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 返回null表示获取锁成功，否则返回锁的剩余生存时间
if (ttl == null) {return;
}// ......// 重试获取锁
while (true) {ttl = tryAcquire(-1, leaseTime, unit, threadId);if (ttl == null) {break;}// ......
}

由以上源码可知，在RedissonLock类的lock()方法中，会调用tryAcquire()方法尝试获取锁。tryAcquire()方法的原理在上一节已经分析过，返回null表示获取锁成功，否则返回锁的剩余生存时间。

如果第一次获取锁失败，程序会进入一个while循环，重试获取锁。

4.5.5 WatchDog机制

// org.redisson.RedissonLockprivate <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {if (leaseTime != -1) {return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);}// 调用Lua脚本RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);ttlRemainingFuture.onComplete((ttlRemaining, e) -> {if (e != null) {return;}// 执行看门狗机制if (ttlRemaining == null) {scheduleExpirationRenewal(threadId);}});return ttlRemainingFuture;
}private void scheduleExpirationRenewal(long threadId) {// ......} else {entry.addThreadId(threadId);// 执行看门狗renewExpiration();}
}private void renewExpiration() {// ......Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {// ......// 调用Lua脚本刷新锁的有效时间RFuture<Boolean> future = renewExpirationAsync(threadId);future.onComplete((res, e) -> {if (e != null) {// loggerreturn;}if (res) {// 递归执行看门狗renewExpiration();}});}// 10s执行一次}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}

由以上源码可知，在RedissonLock类的tryAcquireAsync()方法中，除了调用Lua脚本获取锁，还会运行看门狗机制。该机制会调用Lua脚本刷新锁的有效时间，同时每10s递归执行一次看门狗。

4.5.5 MutiLock原理

为了提高Redis的可用性，一般会搭建集群或者主从。

以主从为例，此时要去获取锁，命令写在主机上，主机会将数据同步给从机。假设在主机还没有来得及把数据写入到从机去的时候，主机宕机了，哨兵会发现主机宕机，并且选举一个Slave变成Master，但此时新的Master中实际上并没有锁信息，相当于此时锁信息已经丢掉了。

为了解决这个问题，Redission提出来了MutiLock锁，使用这种锁后每个节点的地位都是一样的，加锁的逻辑需要把数据写入到每一个主丛节点上，只有所有的节点都写入成功，此时才是真的加锁成功。

假设现在某个节点挂了，那么去获得锁的时候，有一个节点拿不到，不能算是加锁成功，就保证了加锁的可靠性。

4.6 秒杀优化

4.6.1 优化方案

• 现存问题

如上图所示，秒杀下单包括六个步骤：查询优惠券、判断秒杀库存、查询订单、校验一人一单、减库存、创建订单。

在这六步操作中，有很多操作是要去操作数据库的，而且还是一个线程串行执行， 这样就会导致程序执行的很慢。 那么如何加速呢？

• 优化方案

把简单的校验（例如是否有库存、是否一人一单）做完后，就直接给用户返回成功或失败，而不必等待订单创建完成。如果确定可以下单，则将订单的相关信息写入队列，然后再创建一个线程，让新线程读取队列信息异步进行下单。 如下图所示：

• 整体思路

当用户下单时，首先通过Redis判断库存是否充足，如果不充足则直接返回失败；充足的话，再通过Redis判断用户是否已经下过单，如果已经下过单，则直接返回失败；如果没有下过单，则说明可以下单，进行库存扣减，并将用户ID存入当前优惠券的集合中。由于以上过程需要保证原子性，因此可以通过Lua脚本来完成。可以成功下单，Lua脚本返回0。

接着判断Lua脚本的执行结果。如果Lua脚本返回0，说明可以下单，则将优惠券ID、用户ID和订单ID存入阻塞队列，并返回订单ID给用户；如果Lua脚本没有返回0，则直接返回错误信息给用户。

最后进行异步下单，即通过额外线程读取阻塞队列的信息并真正进行下单。完整的流程如下图所示。

4.6.2 完成秒杀优化

• 需求1：新增秒杀优惠券的同时，将优惠券信息保存到Redis中

// com.star.redis.dzdp.service.impl.VoucherServiceImpl@Override
public BaseResult addSeckillVoucher(Voucher voucher) {log.info("add a seckill voucher, {}", voucher.toString());// 1.保存优惠券信息save(voucher);log.info("add voucher success. id = {}", voucher.getId());// 2.保存秒杀信息SeckillVoucher seckillVoucher = new SeckillVoucher();seckillVoucher.setVoucherId(voucher.getId());seckillVoucher.setStock(voucher.getStock());seckillVoucher.setBeginTime(voucher.getBeginTime());seckillVoucher.setEndTime(voucher.getEndTime());seckillVoucherService.save(seckillVoucher);// 3.将秒杀优惠券的库存保存到RedisString key = "seckill:stock:" + voucher.getId();stringRedisTemplate.opsForValue().set(key, voucher.getStock().toString());log.info("set to Redis : Key = {}, Value = {}", key, voucher.getStock().toString());return BaseResult.setOk("新增秒杀券成功！");
}

调用/voucher/seckill/order接口新增一个描述优惠券：

在Redis中可以看到该秒杀优惠券的库存信息：

• 需求2：基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功

在resources目录下新建一个order.lua文件，其内容如下：

-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]-- 2.数据key
-- 2.1.库存key
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2.订单key
local orderKey = 'seckill:order:' .. voucherId-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then-- 3.2.库存不足，返回1return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then-- 3.3.存在，说明是重复下单，返回2return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
return 0

• 需求3：如果抢购成功，将优惠券ID、用户ID和订单ID封装后存入阻塞队列

修改VoucherOrderServiceImpl类的下单方法seckillVoucher()：

// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl/** 保存订单信息的队列 */
private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);@Override
public BaseResult<Long> seckillVoucher(Long voucherId, Long userId) {log.info("开始秒杀下单...voucherId = {}, userId = {}", voucherId, userId);Long orderId = RedisIdWorker.nextId(stringRedisTemplate, "voucher_order");log.info("get orderId = {}", orderId);// 1.执行Lua脚本DefaultRedisScript<Long> script = new DefaultRedisScript<>();script.setLocation(new ClassPathResource("order.lua"));script.setResultType(Long.class);Long result = stringRedisTemplate.execute(script, Collections.emptyList(),voucherId.toString(), userId.toString(), orderId.toString());log.info("execute order.lua result = {}", result);// 2.判断执行结果if(result == null || result != 0) {// 结果为空或者不为0String message = (result == null || result == 1) ? "库存不足" : "不能重复下单";log.error(message);return BaseResult.setFail(message);}// 3.结果为0，将优惠券ID、用户ID和订单ID封装后存入阻塞队列VoucherOrder voucherOrder = new VoucherOrder();voucherOrder.setVoucherId(voucherId);voucherOrder.setUserId(userId);voucherOrder.setId(orderId);orderTasks.add(voucherOrder);log.info("add voucherId = {}, userId = {}, orderId = {} to queue.. done.",voucherId, userId, orderId);// 4.返回订单IDlog.info("秒杀下单返回...orderId = {}", orderId);return BaseResult.setOkWithData(orderId);
}

• 需求4：开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl/** 异步执行下单动作的线程池 */
private static final ExecutorService SECKILL_ORDER_EXECUTOR =Executors.newSingleThreadExecutor();/** 类初始化之后立即初始化线程池 */
@PostConstruct
private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}/*** 处理订单的内部类*/
private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {// while循环持续读取队列中的信息while (true) {try {log.info("=====begin=====>");// 1.获取队列中的订单信息VoucherOrder voucherOrder = orderTasks.take();log.info("get from queue : {}", voucherOrder.toString());// 2.创建订单handleVoucherOrder(voucherOrder);log.info("=====end=====>");} catch (Exception e) {log.error("处理异常订单", e);}}}/*** 处理订单*/private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.创建锁对象RLock rLock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId());// 2.尝试获取锁boolean isLock = rLock.tryLock();log.info("isLock = {}", isLock);// 3.判断是否获取锁成功if(!isLock) {// 获取锁失败log.error("不允许重复下单！");return;}try {// 4.持锁真正创建订单checkAndCreateVoucherOrder(voucherOrder.getVoucherId(), voucherOrder.getUserId());} finally {// 5.释放锁rLock.unlock();log.info("unlock done.");}}/*** 持锁真正创建订单*/private void createVoucherOrder(VoucherOrder voucherOrder) {log.info("begin createVoucherOrder... voucherId = {}, userId = {}, orderId = {}",voucherOrder.getVoucherId(), voucherOrder.getUserId(), voucherOrder.getId());// 1.增加一人一单规则int count = query().eq("voucher_id", voucherOrder.getVoucherId()).eq("user_id", voucherOrder.getUserId()).count();log.info("old order count = {}", count);if(count > 0) {// 该用户已下过单log.error("每个帐号只能抢购一张优惠券！");return;}// 2.扣减库存boolean update = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();log.info("update result = {}", update);if(!update) {// 扣减库存失败，返回抢券失败log.error("库存不足，抢券失败！");return;}// 3.创建订单voucherOrder.setPayTime(new Date());voucherOrderService.save(voucherOrder);}
}

下面借助工具对秒杀下单接口进行性能测试，结果如下：

由于使用的是同一用户，因此971个请求中，只有一个请求是成功的，其余的请求都失败。查看此时Redis中的订单数据，只有1条：

4.7 Redis消息队列

如上图所示，最简单的消息队列包含3个角色：

• 消息队列：存储和管理消息，也被称为消息代理（Message Broker）；
• 生产者：发送消息到消息队列；
• 消费者：从消息队列获取消息并处理消息。

使用队列的好处在于解耦。 在秒杀下单中，用户下单之后，利用Redis去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快了响应速度。

4.7.1 基于List实现消息队列

Redis的List数据结构是一个双向链表，很容易模拟出队列效果。我们可以利用：LPUSH结合RPOP、或者RPUSH结合LPOP来实现。

不过要注意的是，当队列中没有消息时，RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。如图：

基于List的消息队列有哪些优缺点？

优点：

• 利用Redis存储，不受限于JVM内存上限；
• 基于Redis的持久化机制，数据安全性有保证；
• 可以满足消息有序性。

缺点：

• 无法避免消息丢失；
• 只支持单消费者。

4.7.2 基于PubSub的消息队列

PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。如图：

主要命令有：

# 订阅一个或多个频道
SUBSCRIBE channel [channel]
# 订阅与pattern格式匹配的所有频道
PSUBSCRIBE pattern[pattern]
# 向一个频道发送消息
PUBLISH channel msg

基于PubSub的消息队列有哪些优缺点？

优点：

• 采用发布订阅模型，支持多生产、多消费。

缺点：

• 不支持数据持久化；
• 无法避免消息丢失；
• 消息堆积有上限，超出时数据丢失。

4.7.3 基于Stream的消息队列

Stream是Redis 5.0引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

发送消息的命令是：

例如：

127.0.0.1:6379> XADD users * name Rose age 22
"1712458704764-0"
127.0.0.1:6379> XADD users * name Jack age 30
"1712458778623-0"

读取消息的方式之一：XREAD

例如，使用XREAD读取第一个消息：

127.0.0.1:6379> XREAD COUNT 1 STREAMS users 0
1) 1) "users"2) 1) 1) "1712458704764-0"2) 1) "name"2) "Rose"3) "age"4) "22"

XREAD阻塞方式，读取最新消息：

# 阻塞1秒
127.0.0.1:6379> XREAD COUNT 1 BLOCK 1000 STREAMS users $
(nil)
(1.02s)

基于STREAM的消息队列的特点：

• 消息可回溯；
• 一个消息可以被多个消费者读取；
• 可以阻塞读取；
• 有消息漏读的风险。

4.7.4 基于Stream的消息队列-消费者组

消费者组（Consumer Group），就是将多个消费者划分到一个组中，监听同一个队列。它具备下列特点：

创建消费者组：

127.0.0.1:6379> XGROUP CREATE users a_group 0
OK

给自定的消费者组添加消费者：

127.0.0.1:6379> XGROUP CREATECONSUMER users a_group a_consumer1
(integer) 1

从消费者组读取消息：

127.0.0.1:6379> XREADGROUP GROUP a_group a_consumer1 COUNT 1 STREAMS users 0
1) 1) "users"2) (empty array)

基于STREAM消费者组的消息队列的特点：

• 消息可回溯；
• 可以多消费者争抢消息，加快消费速度；
• 可以阻塞读取；
• 没有消息漏读的风险；
• 有消息确认机制，保证消息至少被消费一次。

下面，对比一下这4种消息队列的特点：

经过比较，本案例选择使用基于Stream的消息队列来实现异步秒杀下单。

4.7.5 基于Stream的消息队列实现异步秒杀下单

• 修改秒杀下单Lua脚本order.lua，在认定有抢购资格后，直接向stream.orders队列中添加消息，内容包含voucherId、userId、orderId

-- ...-- 新增逻辑
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)return 0

• 修改消息读取策略，改为读取Redis的Stream结构队列

// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl#seckillVoucher()// ......// 3.结果为0，将优惠券ID、用户ID和订单ID封装后存入阻塞队列
// 新逻辑：这里不再保存队列，在lua脚本中保存
// VoucherOrder voucherOrder = new VoucherOrder();
// voucherOrder.setVoucherId(voucherId);
// voucherOrder.setUserId(userId);
// voucherOrder.setId(orderId);
// orderTasks.add(voucherOrder);
// log.info("add voucherId = {}, userId = {}, orderId = {} to queue.. done.",
//         voucherId, userId, orderId);// 4.返回订单ID
log.info("秒杀下单返回...orderId = {}", orderId);
return BaseResult.setOkWithData(orderId);

// com.star.redis.dzdp.service.impl.VoucherOrderServiceImplprivate class VoucherOrderHandler implements Runnable {@Overridepublic void run() {// 持续读取队列中的信息while (true) {try {log.info("=====begin=====>");// 1.获取队列中的订单信息// VoucherOrder voucherOrder = orderTasks.take();// log.info("get from queue : {}", voucherOrder.toString());// 1.新逻辑：读取Redis的Stream消息队列// XREADGROUP GROUP a_group a_consumer1 COUNT 1 BLOCK 2000 STREAMS stream.orders >List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("a_group", "a_consumer1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create("stream.orders", ReadOffset.lastConsumed()));// 2.判断订单信息是否为空if(list == null || list.isEmpty()) {// 如果为空，说明没有消息，继续下一次循环continue;}// 3.解析数据MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);log.info("get from Redis Stream queue : id = {}, {}", record.getId(), voucherOrder.toString());// 4.创建订单handleVoucherOrder(voucherOrder);// 5.确认消息stringRedisTemplate.opsForStream().acknowledge("stream.orders", "a_group", record.getId());log.info("ack message done.");log.info("=====end=====>");} catch (Exception e) {log.error("处理异常订单", e);}}}// ......
}

测试：

[http-nio-8081-exec-2] 开始秒杀下单...voucherId = 15, userId = 1012
[http-nio-8081-exec-2] get orderId = 7354966481756487681
[http-nio-8081-exec-2] execute order.lua result = 0
[http-nio-8081-exec-2] add voucherId = 15, userId = 1012, orderId = 7354966481756487681 to queue.. done.
[http-nio-8081-exec-2] 秒杀下单返回...orderId = 7354966481756487681
// 创建新线程异步处理下单逻辑
// 成功获取到Stream队列的消息
[pool-2-thread-1] get from Redis Stream queue : id = 1712461578801-0, VoucherOrder(id=7354966481756487681, userId=1012, voucherId=15, payType=null, status=null, createTime=null, payTime=null, useTime=null, refundTime=null, updateTime=null)
[pool-2-thread-1] isLock = true
[pool-2-thread-1] begin createVoucherOrder... voucherId = 15, userId = 1012, orderId = 7354966481756487681
[pool-2-thread-1] ==>  Preparing: SELECT COUNT( * ) FROM tb_voucher_order WHERE (voucher_id = ? AND user_id = ?)
[pool-2-thread-1] ==> Parameters: 15(Long), 1012(Long)
[pool-2-thread-1] <==      Total: 1
[pool-2-thread-1] old order count = 0
[pool-2-thread-1] ==>  Preparing: UPDATE tb_seckill_voucher SET stock = stock - 1 WHERE (voucher_id = ? AND stock > ?)
[pool-2-thread-1] ==> Parameters: 15(Long), 0(Integer)
[pool-2-thread-1] <==    Updates: 1
[pool-2-thread-1] update result = true
[pool-2-thread-1] ==>  Preparing: INSERT INTO tb_voucher_order ( id, user_id, voucher_id, pay_time ) VALUES ( ?, ?, ?, ? )
[pool-2-thread-1] ==> Parameters: 7354966481756487681(Long), 1012(Long), 15(Long), 2024-04-07 11:46:21.208(Timestamp)
[pool-2-thread-1] <==    Updates: 1
[pool-2-thread-1] unlock done.
// 消息确认完成
[pool-2-thread-1] ack message done.

可见，基于Stream的消息队列正常工作。

…

本节完，更多内容请查阅分类专栏：Redis从入门到精通

感兴趣的读者还可以查阅我的另外几个专栏：

• SpringBoot源码解读与原理分析(已完结)
• MyBatis3源码深度解析(已完结)
• 再探Java为面试赋能(持续更新中…)