回顾
我们回顾一下前文下单的流程，当用户发起请求，此时会请求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 若库存不足，则返回1
return 1
end
--3.3 判断用户是否下单
if(redis.call('sismember',orderKey,userId)==1) then
-- 3.4 存在，说明是重复下单，返回2
return 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 {
@Resource
private ISeckillVoucherService seckillVoucherService;
@Resource
private RedisIdWorker redisIdWorker;
@Resource
private StringRedisTemplate stringRedisTemplate;
@Resource
private 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{
@Override
public 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();
}
}
@Transactional
public 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;
@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.判断结果是否为0
int 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.返回订单id
return 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 若库存不足，则返回1
return 1
end
--3.3 判断用户是否下单
if (redis.call('sismember', orderKey, userId) == 1) then
-- 3.4 存在，说明是重复下单，返回2
return 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.判断结果是否为0
int r = result.intValue();
if (r != 0) {
// 2.1.不为0，代表没有购物资格
return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
}
// 3.先获取代理对象，方便处理队列订单时使用
proxy = (IVoucherOrderService) AopContext.currentProxy();
// 4.返回订单id
return 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.判断结果是否为0
int 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.返回订单id
return Result.ok(orderId);
}
*/业务最终代码如下。
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resource
private ISeckillVoucherService seckillVoucherService;
@Resource
private RedisIdWorker redisIdWorker;
@Resource
private StringRedisTemplate stringRedisTemplate;
@Resource
private 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";
@Override
public void run() {
while (true) {
try {
// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders
List<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 id
stringRedisTemplate.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.orders
List<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 id
stringRedisTemplate.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;
@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.判断结果是否为0
int r = result.intValue();
if (r != 0) {
// 2.1.不为0，代表没有购物资格
return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
}
// 3.先获取代理对象，方便处理队列订单时使用
proxy = (IVoucherOrderService) AopContext.currentProxy();
// 4.返回订单id
return Result.ok(orderId);
}
@Transactional
public 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);
}
}
