文章目录
秒杀场景
采用消息队列实现
List实现消息队列
PubSub（发布订阅）实现消息队列
基于Stream实现消息队列
消费者组
实践
总结
秒杀问题是非常重要且比较难实现的，如果不进行架构的优化的话，直接访问会给业务系统造成很大的压力…
秒杀场景
场景一：双十一出现的秒杀场景，比如部分商品开展活动，特价11.11，但是库存一般不多，甚至只有一台。在并发量比较大的情况下，如果我们不进行一些优化的话，很容易出现线程安全问题，而且可能给系统带来太大压力，导致宕机。
如果采用以下的秒杀架构图：
我们可以采用Redis来保证一人一单，我们可以将判断秒杀库存和校验一人一单放到Redis层面来完成，采用异步下单来进行秒杀。
这里我们就需要使用到全面分布式锁中生成的全局ID生成器，因为订单id是需要满足自增、唯一性的条件的，这里也可以采用一些网上的开源方案雪花算法等衍生都可以，但是这样还是会存在问题。什么问题呢？
下单操作的原子性？同步队列的实现？
下单代码是需要校验一人一单和库存等条件的，需要使用lua脚本来保证操作的原子性。
采用阻塞队列来存放数据存在内存限制问题和数据安全问题
采用消息队列实现
消息队列（Message Queue)，也叫存放消息的队列，最简单的消息队列模型包括：
消息队列：存储和管理消息，也叫做消息代理。
生产者：发送消息到消息队列
消费者：从消息队列获取消息并处理消息
在Redis中也提供了三种方式实现消息队列：
List结构：基于List模拟消息队列
PubSub：点对点的消息模型
Stream：比较完善的消息队列模型
List实现消息队列
在Redis中List的数据结构本身就是一个双向链表，所以很容易就可以模拟出队列的效果。
我们可以通过LPUSH结合RPOP、RPUSH结合LPOP实现，不过当队列中没有消息时RPOP或LPOP操作会返回null，并不会像阻塞队列一样等待消息，因此我们也可以使用BRPOP或者BLPOP来实现阻塞的效果。
优点
优点非常明显，不再受限于JVM内存（阻塞队列消费）
采用Redis的持久化机制，数据安全有保证
可以保证消息的有序性
缺点：
无法避免消息丢失
仅支持单消费
PubSub（发布订阅）实现消息队列
消费者可以订阅一个或多个channel，生产者发送对应channel消息后，所以订阅者都能收到相关消息。
乍一看，好像这个确实比List强多了，使用发布订阅模式，还支持多订阅。
但是缺点也很明显：
不支持数据持久化
无法避免消息丢失
消息堆积有限制，超过时也会出现数据丢失
基于Stream实现消息队列
Stream是Redis5.0以后提出的一种数据类型，是功能比较完善的消息队列
发送消息：
读取消息：
除去可以实现读取外，还可以设置阻塞时间，从而实现持续监听的效果
上面的ID指定为$时，标识读取的是最新的消息，如果有同时两条消息同时到达MQ，下次读取可能会出现漏读的问题
消费者组
将多个消费者划分到一个组中，监听同一个队列。
在Redis的Stream的ID策略中
‘>’: 从下一条未消费的消息开始
其它：根据指定id从pending-list中获取已消费但是未确认的消息
这里的确认是因为前面提到可能会丢失消息，而Stream中的消息在经过消费后，需要进行XACK确认，如果没有进行确认就会把这条消息放到pending-list中。
实践
前面提到如何保证秒杀下单的原子性需要使用到lua脚本，在这里面我把对应的key都写死了
-- 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)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0
在消费中，采用ExectorService来创建一个线程池来进行订单的消费
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. 解析消息中的订单信息
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);
// 4. ACK确认 XACK stream.orders g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("处理订单异常");
handlePendingLists();
}
}
}
private void handlePendingLists() {
while (true) {
try {
// 1. 获取pendingList队列中的信息 XREADGROUP GROUP g1 c1 COUNT 1 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. 解析消息中的订单信息
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);
// 4. ACK确认 XACK stream.orders g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("处理pend-list订单异常", e);
try {
Thread.sleep(20);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
}
}
总结
Redis只能满足小项目的需求，更大需求可能需要用到更加高级的消息队列，比如Kafka、RabbitMQ、RocketMQ等等，还有好多学问要学习呢…
