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

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

文章目录

- 前言
- 4.5 分布式锁-Redisson
- - 4.5.4 Redission锁重试
  - 4.5.5 WatchDog机制
  - 4.5.5 MutiLock原理
- 4.6 秒杀优化
- - 4.6.1 优化方案
  - 4.6.2 完成秒杀优化
- 4.7 Redis消息队列
- - 4.7.1 基于List实现消息队列
  - 4.7.2 基于PubSub的消息队列
  - 4.7.3 基于Stream的消息队列
  - 4.7.4 基于Stream的消息队列-消费者组
  - 4.7.5 基于Stream的消息队列实现异步秒杀下单

前言

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.RedissonLock

private <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() {
        @Override
        public void run(Timeout timeout) throws Exception {
            // ......
            // 调用Lua脚本刷新锁的有效时间
            RFuture<Boolean> future = renewExpirationAsync(threadId);
            future.onComplete((res, e) -> {
                if (e != null) {
                    // logger
                    return;
                }
                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.将秒杀优惠券的库存保存到Redis
    String 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.库存不足，返回1
    return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
    -- 3.3.存在，说明是重复下单，返回2
    return 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) {
        // 结果为空或者不为0
        String 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.返回订单ID
    log.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 {
    @Override
    public 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.VoucherOrderServiceImpl

private class VoucherOrderHandler implements Runnable {
    @Override
    public 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从入门到精通

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