1. 基于JVM的阻塞队列的局限
- JVM内存限制问题，大量订单出现时，可能会超过JVM阻塞队列上限；
- 阻塞队列并不能持久化，因为内存不能持久化，出现异常或者宕机之类的故障时，出现数据丢失；
所以引出消息队列的概念
消息队列的两个优点：
- 消息队列在JVM以外的独立服务，不受JVM的内存限制；
- 消息队列不仅仅做数据存储，确保数据安全，会做数据的持久化，并且消费者取数据要做消息确认；如果没有确认，那么消息会在队列中依旧存在，下一次会再投递给消费者，让它继续处理，直到确认为止，确保消息至少消费一次；
2. 基于List结构模拟消息队列
- 假设从队列里取到消息，取到还未处理就发生了异常，这是消息就无法处理了，因为pop相当于remove；
- 发送消息，一旦被消费者拿走之后，别的消费者就无法获得了，无法解决一条消息多个消费者使用；
3. 基于PubSub的消息队列（不建议使用）
PSUBSCRIBE 订阅的格式匹配的三种规则
PubSub消息队列在传递消息时并不会将消息持久化到硬盘上，而是将消息存储在内存中，当服务重启或者发生故障时，可能会导致消息丢失。
4. 基于Stream的消息队列
$:返回最新的消息，前提是该条信息并没有被消费者读过，否则就是返回nil
阻塞等待读取最新的消息，阻塞时间设置为0表示永久等待直到有新等待消息
重点：这种读取方式存在着弊端，当指定其实ID为$时，代表读取最新等待消息，此时处理一条消息的过程中，又来了一条以上的信息到队列，则下次获取也只能获取最新的一条，可能就会出现漏读消息的问题；
当消费者读取一次之后，再生产k4、k5，此时消费者再次阻塞读取最新的消息，再生产消息k6，此时消费者只能获取消息k6，出现了消息漏读；
5. 基于Stream的消息队列问题优化
示例：
消息确认（k1…k5都在pending-list中等待确认）、
查看pending-list中所有未确认的元素
从pendin-list确认未确认的消息，此时消息的起始ID为0
所以可以得出处理消息的大致流程：先利用>的方式去获取所有未消费的消息，然后确认，如果出现异常，在Java中catch采用0的方法去获取pending-list的消息（异常消息），处理完毕再确认，pending-list清空；之后再使用>的方式继续获取未消费的消息，直到阻塞时间过后返回为nil；
6. 基于Redis的Stream结构实现异步秒杀
创建消息队列
修改lua脚本，认定可以抢购直接发送消息到消息队列中
注意：
如果 redis.call('get', stockKey) 返回的结果是空值（nil），那么尝试将空值转换为数字时会出现错误。
因为无法将空值转换为数字。
为了避免这种错误，可以在进行比较之前，先检查返回结果是否为非空值。
这样，如果 redis.call('get', stockKey) 返回的结果是空值，就不会进行比较，从而避免了错误。
-- 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 判断库存是否充足
local stock = redis.call('get', stockKey)
if stock and tonumber(stock) <= 0 then
-- 3.2 库存不足 返回1
return 1
end
-- 3.2 判断用户是否下单
if(redis.call('sismember', orderKey, userId) == 1) then
-- 3.3 存在，说明是重复下单，返回2
return 2
end
-- 3.4 扣库存
redis.call('incrby', stockKey, -1)
-- 3.5 下单（保存用户）
redis.call('sadd', orderKey, userId)
-- 3.6 发送消息到队列中,XDD stream.orders * k1 v1 k2 v2...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0
秒杀的逻辑修改（修改之前的阻塞队列方法）
@Override
public Result seckillVoucher(Long voucherId) {
// 获取用户id
Long userId = UserHolder.getUser().getId();
// 订单id（生成唯一ID）
long orderId = redisIdWorker.nextId("order");
//1. 执行lua脚本(判断购买资格，发送订单信息到消息队列)
Long result = stringRedisTemplate.execute(
SECKILL_SCRIPT,
Collections.emptyList(),// key参数为0，所以参数传空集合
voucherId.toString(),
userId.toString(),
String.valueOf(orderId)
);
//2. 判断是否为0
int i = result.intValue();
if(i != 0) {
//2.1 不为0，没有购买资格
return Result.fail(i == 1 ? "库存不足" : "不能重复下单");
}
//3. 获取代理对象
proxy = (IVoucherOrderService) AopContext.currentProxy();
//4. 返回订单id
return Result.ok(0);
}
开启线程任务
/**
* 异步线程，从消息队列中取出订单信息，执行保存订单到数据库
*/
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.order >
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. 解析消息中的订单信息,键值类型参考脚本中的'userId', userId, 'voucherId', voucherId, 'id', orderId
// 前者String指的是消息ID， <Object, Object>指的是上述键值对
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> values = record.getValue();
// 3.1 转为voucherOrder对象
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
// 4. 如果有，获取成功，可以下单
handleVoucherOrder(voucherOrder);
// 5. ACK确认
// SACK stream.orders g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("订单异常信息",e);
handlePendingList();
}
}
}
/**
* 订单异常消息。采用0的方法去处理pending-list中的消息，进行ACK确认
*/
private void handlePendingList() {
while (true){
try {
// 1. 获取消息队列中的订单信息
// XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.order 0
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 如果获取失败，说明pending-list没有异常消息，继续下一次循环
break;
}
// 3. 解析消息中的订单信息,键值类型参考脚本中的'userId', userId, 'voucherId', voucherId, 'id', orderId
// 前者String指的是消息ID， <Object, Object>指的是上述键值对
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> values = record.getValue();
// 3.1 转为voucherOrder对象
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
// 4. 如果有，获取成功，可以下单
handleVoucherOrder(voucherOrder);
// 5. ACK确认
// SACK stream.orders g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("订单异常信息",e);
try {
Thread.sleep(20);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}
}
}
}
