秒杀优化(异步秒杀，基于redis-stream实现消息队列)

秒杀优化

一：异步秒杀

1：思路

原本我们每一个请求都是串行执行，从头到尾执行完了才算一个请求处理成功，这样过于耗时，我们看到执行的操作中查询优惠券，查询订单，减库存，创建订单都是数据库操作，而数据库的性能又不是很好，我们可以将服务拆分成两部分，将判断优惠券信息和校验一人一单的操作提取出来，先执行判断优惠券和校验操作，然后直接返回订单id，我们在陆续操作数据库减库存和创建订单，这样前端响应的会非常快，并且我们可以将优惠券和一人一单的操作放在redis中去执行，这样又能提高性能，然后我们将优惠券信息，用户信息，订单信息，先保存在队列里，先返回给前端数据，在慢慢的根据队列的信息去存入数据

我们之前说将查询和校验功能放在redis中实现，那么用什么结构呢，查询订单很简单，只要查询相应的优惠券的库存是否大于0就行，我们就可以是否字符串结构，key存优惠券信息，value存库存；那么校验呢，因为是一人一单，所以我们可以使用set，这样就能保证用户的唯一性；

我们执行的具体步骤是：先判断库存是否充足，不充足直接返回，充足判断是否有资格购买，没有返回，有就可以减库存，然后将用户加入集合中，在返回，因为我们执行这些操作时要保证命令的原子性，所以这些操作我们都使用lua脚本来编写；

具体的执行流程就是，先执行lua脚本，如果结果不是0那么直接返回，如果不是0，那么就将信息存入阻塞队列然后返回订单id；

2：实现

1：新增时添加到redis

stringRedisTemplate.opsForValue().set(RedisConstants.SECKILL_STOCK_KEY+voucher.getId(),voucher.getStock().toString());

2：lua脚本编写:

local stock =tonumber(redis.call('get', 'seckill:stock:' .. ARGV[1]))
if (stock<=0) thenreturn 1
end
local userId=ARGV[2]
local isok=tonumber(redis.call('sadd','seckill:order:'..ARGV[1],userId))
if isok==0 thenreturn 2
end
redis.call('incrby','seckill:stock:'..ARGV[1],-1)
return 0

然后就能改变之前的代码，在redis中实现异步下单：

@Override
public Result seckilOrder(Long voucherId) throws InterruptedException {Long id = UserHolder.getUser().getId();Long res = (Long) stringRedisTemplate.execute(SECKIL_ORDER_LUA, Collections.emptyList(), voucherId.toString(), id.toString());if (res!=0){return Result.fail(res==1?"库存不足":"一人只能购买一单");}long orderID = redisIDWork.nextId("order");return Result.ok(orderID);}

初始化lua脚本文件

@Resource
private RedissonClient redissonClient2;
public static final DefaultRedisScript SECKIL_ORDER_LUA;
static {//初始化SECKIL_ORDER_LUA=new DefaultRedisScript<>();//定位到lua脚本的位置SECKIL_ORDER_LUA.setLocation(new ClassPathResource("seckill.lua"));//设置lua脚本的返回值SECKIL_ORDER_LUA.setResultType(Long.class);
}

还剩一个阻塞队列没有实现：

阻塞队列的功能就是异步的将订单信息存入数据库；

阻塞队列可以使用blockdeque

BlockingQueue<VoucherOrder> blockingQueue = new ArrayBlockingQueue<VoucherOrder>(1024*1024);

在类上直接初始化

然后使用的时候就是，将订单添加到阻塞队列，让另一个线程去执行，往数据库中添加阻塞队列中的订单信息：

blockingQueue.add(voucherOrder);

然后就要开出一个线程，然后执行往数据库添加元素的任务了：

 //创建一个线程private ExecutorService SECKILL_ORDER_EXECUTOR=Executors.newSingleThreadExecutor();//注解PostConstruct，添加这个注解的方法就是在类初始化完成之后就会执行；@PostConstructprivate void init(){//提交任务SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandle());}//定义一个任务内部类，实现Runnable，然后需要实现run方法，run方法中就是我们的任务private class VoucherOrderHandle implements Runnable {@Overridepublic void run() {try {//从阻塞队列中取出订单VoucherOrder voucherOrder = blockingQueue.take();//执行方法handleVoucherOrder(voucherOrder);} catch (InterruptedException e) {log.info("下单业务异常",e);}}}

当类加载是就会一直提交任务，只要阻塞队列里有订单，就会将订单取出然后调用方法将订单存入数据库

调用的方法是尝试获取锁的方法，而获取锁其实并不需要，因为我们自己开出来的线程只有一个是单线程，而且在lua脚本中已经对一人一单还有超卖问题进行处理，这里只是为了更加保险

 @Transactionalpublic void handleVoucherOrder(VoucherOrder voucherOrder) throws InterruptedException {
//        SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order" + id, stringRedisTemplate);Long userId = voucherOrder.getUserId();RLock simpleRedisLock = redissonClient2.getLock("lock:order" + userId);boolean trylock = simpleRedisLock.tryLock(1L, TimeUnit.SECONDS);if (!trylock){log.info("获取锁失败");}try {orderService.createVoucherOrder(voucherOrder);} catch (IllegalStateException e) {throw new RuntimeException(e);}finally {simpleRedisLock.unlock();}}

然后获取锁成功后就会调用方法执行数据库操作，但是这个方法是带有事务的，我们单独开出来的子线程无法使事务生效，只能在方法的外部声明一个代理对象，然后通过代理对象去调用方法使事务生效；

 @Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {Integer count = query().eq("user_id", voucherOrder.getUserId()).eq("voucher_id", voucherOrder.getVoucherId()).count();if (count > 0) {log.info("一个用户只能下一单");}//进行更新，库存减一boolean success = seckillVoucherService.update().setSql("stock = stock - 1") // set stock = stock - 1.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();// where id = ? and stock > 0//扣减失败，返回错误信息；if (!success) {log.info("扣减失败");}save(voucherOrder);}

因为我们是开出来的子线程调用的方法，所以不能从线程中获取值，只能从我们传入的订单对象获取，然后就是减库存和存入订单的操作了；

总结：

我们使用异步操作，将下单和存入订单分开来执行，大大提高了执行的销量，在redis中完成超卖和一人一单的问题；

然后使用阻塞队列，开出一个子线程异步存入数据库下单；

问题：

我们的阻塞队列是在jvm中的，jvm中内存是有上线的，超过上限就会有异常，还有就是我们的数据都是存放在内存中，要是出现了一些事故会导致数据丢失

二：redis实现消息队列

1：什么是消息队列

消息队列由三个角色构成：

1：生产者：发送消息到消息队列

2：消息队列：存储和管理消息队列，也被称为消息代理

3：消费者：从消息队列中获取消息并处理

好的消息队列有这几个特点：

1：有独立的服务，独立的内存；

2：可以做到数据的持久化

3：能够发送消息给消费者并且确保消息处理完成

2：基于list结构实现消息队列

使用brpop可以实现阻塞获取

3：基于pubsub实现消息队列

4：基于stream实现消息队列

stream发送消息的方式xadd key * msg

key是指消息队列的名称，* 是发送消息的名称由redis来生成，后面的msg就是键值对，我们要发宋的消息

xread是读取消息的命令：count指定读取消息的数量，block指定阻塞时间，不指定就是不阻塞，指定0就是无限等待，sreams 是消息队列的名称，可以是多个，id是消息的id，0是从0开始读，$是从最新的开始读

但是有个问题就是，指定$是获取最新的消息，但是只是获取使用这个命令之后最新的消息，而如果一次性发多条，只会获取最后一个，就会出现漏消息；

5：stream的消费者组模式

消费者组就是将消费者划分到一个组中监听一个消息队列：

有这些好处：

1：消息分流：消息发送到消费者组中，消费者会处于竞争关系，会争夺消息来处理，这个发送多个消息就会实现分流，就会由不同的消费者来处理，加快了处理速度；

2：消息标识：在读取消息后会记录最后一个被处理的消息，这样就不会出现消息漏读的情况；

3：消息确认：消息发出去会，消息会处于pending状态，会等待消息处理完毕，这个时候会将消息存入pendinglist中，当处理完后才会从pending中移除；确保了消息的安全性，保证消息不会丢失，就算再消息发出去后，服务宕机了，也能知道该消息没有被处理，这个功能的作用就是确保消息至少被消费一次；

三：基于redis的stream结构实现消息队列

首先再redis客户端中输入命令创建一个队列和接受这个队列消息的组

然后修改秒杀下单的lua脚本，直接在redis中通过消息队列将消息发送给消费者：

local orderId=ARGV[3]
local stock =tonumber(redis.call('get', 'seckill:stock:' .. ARGV[1]))
if (stock<=0) thenreturn 1
end
local userId=ARGV[2]
local isok=tonumber(redis.call('sadd','seckill:order:'..ARGV[1],userId))
if isok==0 thenreturn 2
end
redis.call('incrby','seckill:stock:'..ARGV[1],-1)
--将消息发送给stream.orders队列
redis.call('xadd','stream.orders','*','userId',userId,'id',orderId,'voucherId',ARGV[1])
return 0

这里发送的是优惠券id，用户id还有订单id，正是我们存入数据库中所需要的参数

然后就可以去修改前面秒杀下单的逻辑，不用去将消息放到阻塞队列，我们直接从redis的队列中取出就行；

@Override
public Result seckilOrder(Long voucherId) throws InterruptedException {long orderId = redisIDWork.nextId("order");Long userId = UserHolder.getUser().getId();Long res = (Long) stringRedisTemplate.execute(SECKIL_ORDER_LUA,Collections.emptyList(), voucherId.toString(),userId.toString(),String.valueOf(orderId));if (res != 0) {return Result.fail(res == 1 ? "库存不足" : "一人只能购买一单");}orderService = (IVoucherOrderService) AopContext.currentProxy();return Result.ok(orderId);
}

这里我们需要将订单id作为lua脚本的参数传入进去，然后将订单信息存入阻塞队列的操作可以省略，因为我们已经将订单信息存入了redis中的消息队列；

然后这里我们需要单独开出一个线程去将队列中的消息存入数据库：

private class VoucherOrderHandle implements Runnable {String ququeName="stream.orders";@Overridepublic void run() {try {//从消息队列中取出订单while (true){//xreadgroup GROUP group consumer count(1) block(2000) streams key  >List<MapRecord<String, Object, Object>> msg = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)), StreamOffset.create(ququeName, ReadOffset.lastConsumed()));//如果消息为空就继续等待接收if (msg==null||msg.isEmpty()){continue;}//因为每次读取一个消息，所以我们获取第一个消息MapRecord<String, Object, Object> entries = msg.get(0);//获取消息的值，是一些我们传入的键值对Map<Object, Object> value = entries.getValue();//将map转成voucherorder对象VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value,new VoucherOrder(),false);//执行方法handleVoucherOrder(voucherOrder);//确认消息已经处理stringRedisTemplate.opsForStream().acknowledge(ququeName,"g1",entries.getId());}} catch (InterruptedException e) {log.info("下单业务异常",e);handleVoucherOrderError();}}

我们要做的就是接受消息，然后再将消息存入数据库：

我们调用stream的方法，作为消费者从队列中读取消息，阻塞时间是2秒，每次读取一个消息，从下一个未消费的消息读取，如果读取的消息为空那么就继续循环读取消息，如果有消息就将消息取出，然后将其转成对象map，再将其转成对象，然后再去做确认消息的处理，如果不确认消息，消息就会存在待处理的队列中；如果出现的异常，那么我们取出的消息可能没有进行确认，没有确认的会存入待处理队列，我们就要从队列里取出然后进行处理；

出错只会执行的方法：

 private void handleVoucherOrderError() {try {//从消息队列中取出订单while (true){//xreadgroup GROUP group consumer count(1)  streams key  0,表示从第一个未处理的消息开始读取List<MapRecord<String, Object, Object>> msg = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1), StreamOffset.create(ququeName, ReadOffset.from("0")));//如果为空就说明没有待处理的消息结束就行if (msg==null||msg.isEmpty()){break;}//因为每次读取一个消息，所以我们获取第一个消息MapRecord<String, Object, Object> entries = msg.get(0);//获取消息的值，是一些我们传入的键值对Map<Object, Object> value = entries.getValue();//将map转成voucherorder对象VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value,new VoucherOrder(),false);//执行方法handleVoucherOrder(voucherOrder);}} catch (InterruptedException e) {log.info("下单业务异常",e);}}
}

这里因为是再待处理中直接取出，所以不用阻塞处理，然后从待消费队列中第一个消息开始读，如果为空，那么就说明没有待处理的消息，我们直接返回就行，如果不为空我们再处理

这样使用redis中的消息队列就实现了：1：独立的服务，足够的内存；2：有确认机制，避免消息漏读；3：消息持久化

BeanUtil.fillBeanWithMap(value,new VoucherOrder(),false);
//执行方法
handleVoucherOrder(voucherOrder);
}
} catch (InterruptedException e) {
log.info(“下单业务异常”,e);
}
}
}

> 这里因为是再待处理中直接取出，所以不用阻塞处理，然后从待消费队列中第一个消息开始读，如果为空，那么就说明没有待处理的消息，我们直接返回就行，如果不为空我们再处理这样使用redis中的消息队列就实现了：1：独立的服务，足够的内存；2：有确认机制，避免消息漏读；3：消息持久化![外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传](https://i-blog.csdnimg.cn/direct/9d63425d09764b7c8b385a64615924a2.png)