Redis实战—秒杀优化（Redis消息队列）

回顾

基于阻塞队列实现程序异步优化

优化代码

总结

Redis消息队列

初始消息队列

基于List结构模拟消息队列

基于PubSub的消息队列

基于Stream的消息队列

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

总结

代码实现

回顾

        我们回顾一下前文下单的流程，当用户发起请求，此时会请求nginx，nginx会访问到tomcat，而tomcat中的程序，会进行串行操作，分成如下几个步骤。
        1、查询优惠卷
        2、判断秒杀库存是否足够
        3、查询订单
        4、校验是否满足一人一单
        5、扣减库存
        6、创建订单
        在这六步操作中，有很多操作是要去操作数据库的，而且还是一个线程串行执行，这样就会导致我们的程序执行的很慢，所以我们需要异步程序执行，那么该如何优化呢？

基于阻塞队列实现程序异步优化

优化方案：我们将耗时较短的逻辑判断交给redis操作，比如库存是否足够，是否满足一人一单的条件，只要满足这两项判断，就意味着我们一定可以完成下单，因此，我们只需要快速进行逻辑判断，根本不用等下单逻辑走完，就可以先给用户发送响应信息。若用户可以下单，再在后台开一个线程，让后台线程慢慢去执行queue里边的消息，这样程序耗时将大大缩短，而且也不用担心线程池消耗殆尽的问题，因为我们的程序中并没有手动使用任何线程池。

在用户下单之后，判断库存是否充足只需要到redis中找到对应优惠券key的value（后台在添加秒杀优惠券时，会对应将该优惠券及库存添加到redis中），并判断其是否大于0即可。如果不充足，则直接结束；如果充足，则继续在redis中判断该用户是否已经下过单，如果set集合中不存在该用户的记录，则说明该用户从未下过单，满足下单条件。整个过程需要保证原子性，我们可以使用lua来操作。

优化代码

1. 首先，在添加优惠券的同时，我们需要将该优惠券及其库存保存到redis中，方便我们之后在redis中快速判断优惠券库存是否充足。对添加优惠券方法做修改如下。

@Override
@Transactional
public void addSeckillVoucher(Voucher voucher) {// MP保存优惠券save(voucher);// 保存秒杀信息SeckillVoucher seckillVoucher = new SeckillVoucher();seckillVoucher.setVoucherId(voucher.getId());seckillVoucher.setStock(voucher.getStock());seckillVoucher.setBeginTime(voucher.getBeginTime());seckillVoucher.setEndTime(voucher.getEndTime());seckillVoucherService.save(seckillVoucher);// 保存秒杀库存到Redis中stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
}

2. redis判断采用lua脚本，代码如下。

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

3. 基于阻塞队列实现秒杀优化

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//初始化返回值SECKILL_SCRIPT.setResultType(Long.class);}@PostConstruct //注解含义：在当前类初始化完毕后执行private void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}//创建阻塞队列private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024);//创建线程池private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();//创建线程任务private class VoucherOrderHandler implements Runnable{@Overridepublic void run() {while (true){try {// 1.获取队列中的订单信息VoucherOrder voucherOrder = orderTasks.take();// 2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("处理订单异常",e);}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.获取用户Long userId = voucherOrder.getUserId();// 2.创建锁对象RLock lock = redissonClient.getLock("lock:order:" + userId);// 3.获取锁boolean isLock = lock.tryLock();// 4.若获取锁失败if (!isLock) {log.error("不允许重复下单");return;}// 获取锁成功 （理论上没有问题，lua脚本已经判断过了，这里再加锁只是兜底）try {//通过代理对象调用proxy.createVoucherOrder(voucherOrder);} finally {lock.unlock();}}@Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {boolean success = seckillVoucherService.update().setSql("stock = stock - 1")  //使用MP，设置sql语句.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();save(voucherOrder);}private IVoucherOrderService proxy;@Overridepublic Result seckillVoucher(Long voucherId) {// 获取用户Long userId = UserHolder.getUser().getId();// 1.执行lua脚本，判断用户是否用购买资格（库存不足与重复下单问题）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());// 2.判断结果是否为0int r = result.intValue();if(r!=0){// 2.1.不为0，代表没有购物资格return Result.fail(r==1?"库存不足":"不能重复下单");}// 2.2 为0，有购买资格，先创建订单，再将订单信息添加到阻塞队列VoucherOrder voucherOrder = new VoucherOrder();// 2.3 获取订单id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);voucherOrder.setVoucherId(voucherId);// 2.4将订单信息存入阻塞队列，任务结束orderTasks.add(voucherOrder);//3.获取代理对象，方便后序线程使用，可以放在成员变量或者是voucherOrder里面proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回订单idreturn Result.ok(orderId);}}

总结

秒杀业务的优化思路是什么？
1.先利用Redis完成库存余量、一人一单判断，创建抢单业务
2.再将抢单业务放入阻塞队列，利用独立线程异步下单

基于阻塞队列的异步秒杀存在哪些问题？
1.内存限制问题
2.数据安全问题

Redis消息队列

初始消息队列

消息队列就是存放消息的队列，最简单的消息队列模型包括以下3个角色。
1.消息队列：存储和管理消息，也被称为消息代理（Message Broker）
2.生产者：发送消息到消息队列
3.消费者：从消息队列获取消息并处理消息

使用消息队列的好处
消息队列能够将我们的业务进行解耦，举一个生活中的例子就是：快递员(生产者)把快递放到快递柜(Message Queue)里边，我们(消费者)再从快递柜里拿东西，这就是一个异步。如果耦合，相当于快递员直接把快递交给你，这事固然好，但是万一你不在家，那么快递员就会一直等你，这便浪费了快递员的时间，所以这种思想在我们日常开发中，是非常有必要的。

基于List结构模拟消息队列

消息队列（Message Queue），字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表，很容易模拟出队列效果。

队列是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。

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

优点：
1.利用Redis存储，不受限于JVM内存上限
2.基于Redis的持久化机制，数据安全性有保证
3.可以满足消息有序性

缺点：
1.无法避免消息丢失
2.只支持单消费者

基于PubSub的消息队列

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

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

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

优点：
1.采用发布订阅模型，支持多生产、多消费

缺点：
1.不支持数据持久化
2.无法避免消息丢失
3.消息堆积有上限，超出时数据丢失

基于Stream的消息队列

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

发送消息命令如下。

举例如下。

读取消息的方式之一：XREAD

举例如下。
XREAD阻塞方式，读取最新消息如下。
在业务开发中，我们可以循环的调用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下。
STREAM类型消息队列的XREAD命令特点
1.消息可回溯
2.一个消息可以被多个消费者读取
3.可以阻塞读取
4.有消息漏读的风险

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

常用命令如下。

XGROUP CREATE key groupName ID [MKSTREAM]
key：队列名称
groupName：消费者组名称
ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
MKSTREAM：队列不存在时自动创建队列删除指定的消费者组
XGROUP DESTORY key groupName给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupname consumername删除消费者组中的指定消费者
XGROUP DELCONSUMER key groupname consumername从消费者组读取消息：
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
group：消费组名称
consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
count：本次查询的最大数量
BLOCK milliseconds：当没有消息时最长等待时间
NOACK：无需手动ACK，获取到消息后自动确认
STREAMS key：指定队列名称
ID：获取消息的起始ID：">"：从下一个未消费的消息开始其它：根据指定id从pending-list中获取已消费但未确认的消息。例如0，是从pending-list中的第一个消息开始

STREAM类型消息队列的XREADGROUP命令特点
1.消息可回溯
2.可以多消费者争抢消息，加快消费速度
3.可以阻塞读取
4.没有消息漏读的风险
5.有消息确认机制，保证消息至少被消费一次

消费者监听消息的基本思路如下图所示。

总结

代码实现

基于Redis的Stream结构作为消息队列，实现异步秒杀下单，需求如下。
1.创建一个Stream类型的消息队列，名为stream.orders
2.修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包3.含voucherId、userId、orderId
4.项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

首先创建消息队列。

XGROUP CREATE stream.orders  g1 0 MKSTREAM

其次，修改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 判断库存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0) then--3.2 若库存不足，则返回1return 1
end--3.3 判断用户是否下单
if (redis.call('sismember', orderKey, userId) == 1) then-- 3.4 存在，说明是重复下单，返回2return 2
end--3.5 扣库存
redis.call('incrby', stockKey, -1)
--3.6 下单（保存用户下单记录）
redis.call('sadd', orderKey, userId)
--3.7 发送消息到队列中，XADD stream.order * k1 v1 k2 v2
redis.call('xadd', 'stream.orders','*','userId',userId,'voucherId',voucherId,'id',orderId)
return 0

改造秒杀业务代码如下。

@Override
public Result seckillVoucher(Long voucherId) {// 获取用户Long userId = UserHolder.getUser().getId();// 获取订单id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");// 1.执行lua脚本，判断用户是否用购买资格（库存不足与重复下单问题）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));// 2.判断结果是否为0int r = result.intValue();if (r != 0) {// 2.1.不为0，代表没有购物资格return Result.fail(r == 1 ? "库存不足" : "不能重复下单");}// 3.先获取代理对象，方便处理队列订单时使用proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回订单idreturn Result.ok(orderId);
}/*未修改前的代码如下
@Override
public Result seckillVoucher(Long voucherId) {// 获取用户Long userId = UserHolder.getUser().getId();// 1.执行lua脚本，判断用户是否用购买资格（库存不足与重复下单问题）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());// 2.判断结果是否为0int r = result.intValue();if(r!=0){// 2.1.不为0，代表没有购物资格return Result.fail(r==1?"库存不足":"不能重复下单");}// 2.2 为0，有购买资格，先创建订单，再将订单信息添加到阻塞队列VoucherOrder voucherOrder = new VoucherOrder();// 2.3 获取订单id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);voucherOrder.setVoucherId(voucherId);// 2.4将订单信息存入阻塞队列，任务结束orderTasks.add(voucherOrder);// 3.先获取代理对象，方便处理队列订单时使用proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回订单idreturn Result.ok(orderId);
}
*/

业务最终代码如下。

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//初始化返回值SECKILL_SCRIPT.setResultType(Long.class);}@PostConstruct //注解含义：在当前类初始化完毕后执行private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}//创建阻塞队列private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);//创建线程池private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();//创建线程任务private class VoucherOrderHandler implements Runnable {String queueName = "stream.orders";@Overridepublic void run() {while (true) {try {// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.ordersList<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(queueName, ReadOffset.lastConsumed()));// 2.判断消息获取是否成功if (list == null || list.isEmpty()) {// 2.1.如果获取失败，说明没有消息，继续下一次循环continue;}// 3.解析消息中的订单信息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> values = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);// 3.如果获取成功，可以下单handleVoucherOrder(voucherOrder);// 5.ACK确认 SACK stream.orders g1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());} catch (Exception e) {log.error("处理订单异常", e);handlePendingList();}}}private void handlePendingList() {while (true) {try {// 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.ordersList<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1),StreamOffset.create(queueName, ReadOffset.from("0")));// 2.判断消息获取是否成功if (list == null || list.isEmpty()) {// 2.1.如果获取失败，说明没有消息，跳出循环break;}// 3.解析消息中的订单信息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> values = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);// 3.如果获取成功，可以下单handleVoucherOrder(voucherOrder);// 5.ACK确认 SACK stream.orders g1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("处理pending-list订单异常", e);try {Thread.sleep(20);} catch (InterruptedException ex) {ex.printStackTrace();}}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.获取用户Long userId = voucherOrder.getUserId();// 2.创建锁对象RLock lock = redissonClient.getLock("lock:order:" + userId);// 3.获取锁boolean isLock = lock.tryLock();// 4.若获取锁失败if (!isLock) {log.error("不允许重复下单");return;}// 获取锁成功try {//通过代理对象调用proxy.createVoucherOrder(voucherOrder);} finally {lock.unlock();}}private IVoucherOrderService proxy;@Overridepublic Result seckillVoucher(Long voucherId) {// 获取用户Long userId = UserHolder.getUser().getId();// 获取订单id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");// 1.执行lua脚本，判断用户是否用购买资格（库存不足与重复下单问题）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));// 2.判断结果是否为0int r = result.intValue();if (r != 0) {// 2.1.不为0，代表没有购物资格return Result.fail(r == 1 ? "库存不足" : "不能重复下单");}// 3.先获取代理对象，方便处理队列订单时使用proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回订单idreturn Result.ok(orderId);}@Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {boolean success = seckillVoucherService.update().setSql("stock = stock - 1")  //使用MP，设置sql语句.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();save(voucherOrder);}
}