目录
1、概述
1、消息队列（Message Queue），字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：
- 消息队列：存储和管理消息，也被称为消息代理（Message Broker）
- 生产者：发送消息到消息队列
- 消费者：从消息队列获取消息并处理消息
2、Redis提供了三种不同的方式来实现消息队列：
- list结构：基于List结构模拟消息队列
- 优点：
- 利用Redis存储，不受限于JVM内存上限
- 基于Redis的持久化机制，数据安全性有保证
- 可以满足消息有序性
- 缺点：
- 无法避免消息丢失
- 只支持单消费者
- PubSub：发布订阅消息模型
- 优点：
- 采用发布订阅模型，支持多生产、多消费
- 缺点：
- 不支持数据持久化
- 无法避免消息丢失
- 消息堆积有上限，超出时数据丢失
- Stream：比较完善的消息队列模型
2、基于Stream的消息队列-单消费者
1、命令
Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。
发送消息的命令：
读取消息的命令：
2、Stream类型的消息队列的XREAD命令特点：
- 消息可回溯（即可重复读取）
- 一个消息可以被多个消费者读取
- 可以阻塞读取
- 有消息漏读的风险
3、基于Stream的消息队列-消费者组
消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：
- 创建消费者组：
key：队列名称
groupName：消费者组名称
ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
MKSTREAM：队列不存在时自动创建队列
- 从消费者组读取消息：
group：消费组名称
consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
count：本次查询的最大数量
BLOCK milliseconds：当没有消息时最长等待时间
NOACK：无需手动ACK，获取到消息后自动确认
STREAMS key：指定队列名称
ID：获取消息的起始ID：
">"：从下一个未消费的消息开始
其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从list中的第一个消息开始
- Stream类型的消息队列的XREADGROUP命令特点：
- 消息可回溯
- 可以多消费者争抢消息，加快消费速度
- 可以阻塞读取
- 没有消息漏读的风险
- 有消息确认机制，保证消息至少被消费一次
4、基于Stream的消息队列实现异步秒杀
1、实现思路
- 创建一个Stream类型的消息队列，名为stream.orders
- 修改之前的秒杀下单Lua脚本，判断有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId
- 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单
2、代码实现
--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,这里的value为下单的用户ID
local orderKey = 'seckill:order:' .. voucherId
--3.脚本业务
--3.1判断库存是否充足 get stockKey
if(tonumber(redis.call('get',stockKey)) <= 0) then
--库存不足，返回1
return 1
end
--3.2判断用户是否已经购买过 sismember orderKey userId
if(redis.call('sismember',orderKey,userId) == 1) then
--存在，则该用户重复下单，返回2
return 2
end
--3.3扣减库存 incrby stockKey -1
redis.call('incrby',stockKey,-1)
--3.4生成订单(将用户id添加到订单的set集合中)
redis.call('sadd',orderKey,userId)
--3.5将订单信息发送到消息队列中 XADD stream.orders * k1 v1 k2 v2 k3 v3
redis.call('xadd','stream.orders','*','userId',userId,'voucherId',voucherId,'id',orderId)
return 0/**
* 优惠券订单模块
*/
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Autowired
private ISeckillVoucherService seckillVoucherService;
@Autowired
private RedisIDWorker redisIDWorker;
@Autowired
private StringRedisTemplate stringRedisTemplate;
//初始化脚本对象,用于加载Lua脚本 这里定义为static是为了在类加载时就初始化该脚本对象，并且只会初始化一次，提高IO性能
private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
SECKILL_SCRIPT = new DefaultRedisScript<>();
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
SECKILL_SCRIPT.setResultType(Long.class);
}
//创建单线程的线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
//在当前类初始化完毕后就执行该方法
@PostConstruct
private void init(){
SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}
//创建线程任务--从消息队列中读取订单信息并下单
private class VoucherOrderHandler implements Runnable{
//消息队列名称
private String queueName = "stream.orders";
@Override
public void run() {
while (true){
try {
//从消息队列中读取消息 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())
);
//判断是否读取到消息
if (list == null || list.isEmpty()) {
//读取失败，说明没有消息，重新读取
continue;
}
//读取成功，解析消息中的订单信息
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> map = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(map, new VoucherOrder(),true);
//下单
handleVoucherOrder(voucherOrder);
//ACK确认 XACK stream.orders g1 消息id
stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",record.getId());
} catch (Exception e) {
log.error("处理订单异常",e);
//处理消息时出现异常，则读取pending-list中的未确认的消息
handlePendingList();
}
}
}
//读取pending-list中已读取但未确认的消息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders 0
private void handlePendingList() {
while (true){
try {
//从pending-list中读取消息
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"),
StreamReadOptions.empty().count(1),
StreamOffset.create(queueName, ReadOffset.lastConsumed())
);
//判断是否读取到消息
if (list == null || list.isEmpty()) {
//读取失败，说明pending-list中没有异常消息，结束循环
break;
}
//读取成功，解析消息中的订单信息
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> map = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(map, new VoucherOrder(),true);
//下单
handleVoucherOrder(voucherOrder);
//ACK确认
stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",record.getId());
} catch (Exception e) {
log.error("处理pending-list订单异常",e);
}
}
}
}
//创建代理对象
private IVoucherOrderService proxy;
/**
* 创建订单到数据库,独立线程会执行此方法
* @param voucherOrder
*/
public void handleVoucherOrder(VoucherOrder voucherOrder){
//获取用户ID 这里不能使用ThreadLocal获取用户信息，因为此时是新的独立线程在执行该方法
Long userId = voucherOrder.getUserId();
//创建锁对象
SimpleRedisLock redisLock = new SimpleRedisLock("order"+userId.toString(),stringRedisTemplate);
//尝试获取锁
boolean isLock = redisLock.tryLock(2);
//判断是否获取到
if (!isLock){
//获取锁失败，返回错误
log.error("不允许重复下单");
return;
}
try {
//创建订单到数据库
proxy.createVoucherOrder(voucherOrder);
} finally {
//释放锁
redisLock.unLock();
}
}
/**
* 优惠券秒杀下单功能：基于Redis消息队列实现异步秒杀
* @param voucherId
* @return
*/
@Override
public Result seckillVoucher(Long voucherId) {
//获取当前用户
Long userId = UserHolder.getUser().getId();
//订单id
long orderId = redisIDWorker.nextId("order");
//执行Lua脚本
Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
Collections.emptyList(),
voucherId.toString(), userId.toString(),String.valueOf(orderId));
//判断结果是否为0
int r = result.intValue();
if (r != 0){
//不为0，返回异常信息
return Result.fail(r==1 ? "库存不足" : "不能重复购买");
}
//为0，则有购买资格
//获取代理对象（事务） 保证事务提交之后再释放锁
proxy = (IVoucherOrderService) AopContext.currentProxy();
//返回订单id给前端
return Result.ok(orderId);
}
/**
* 创建优惠券订单
* @param voucherOrder
*/
@Transactional
public void createVoucherOrder(VoucherOrder voucherOrder) {
//实现一人一单
Long userId = voucherOrder.getUserId();
//查询优惠券订单
int count = query().eq("voucher_id", voucherOrder.getVoucherId()).eq("user_id", userId).count();
//判断订单是否存在
if (count>0){
//该用户已经购买过了
log.error("不能重复购买");
return;
}
//扣减库存 update tb_seckill_voucher set stock=stock-1 where voucher_id=? and stock>0
boolean success = seckillVoucherService.update()
.setSql("stock = stock -1")
.eq("voucher_id", voucherOrder.getVoucherId())
.gt("stock",0)
.update();
//判断扣减是否成功
if (!success){
log.error("库存不足");
return;
}
//新增订单
save(voucherOrder);
}
}3、总结
通过Redis的Stream类型消息队列实现异步秒杀的功能，利用消费者组的方式将多个消费者划分为同一个组，监听同一个队列，利用消息分流提高了消息消费的速度，减少消息堆积；利用消息标示确保每个消息都能被消费，解决了消息遗漏的问题；利用消息确认机制保证了每个消息都能至少被处理一次，成功地解决了基于阻塞队列带来的内存限制问题和数据安全问题。
本文含有隐藏内容，请 开通VIP 后查看