当前位置： 首页 > news >正文

Redis--day10--黑马点评--秒杀优化消息队列

news 2025/8/24 8:30:38

请添加图片描述
（以下内容全部来自上述课程）
在这里插入图片描述

秒杀优化

请添加图片描述
请添加图片描述
改进秒杀业务，提高并发性能
需求:

  1. 新增秒杀优惠券的同时，将优惠券信息保存到Redis中
package com.hmdp.service.impl;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.Voucher;
import com.hmdp.mapper.VoucherMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherService;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;
import java.util.List;import static com.hmdp.utils.RedisConstants.SECKILL_STOCK_KEY;/*** <p>*  服务实现类* </p>** @author 虎哥* @since 2021-12-22*/
@Service
public class VoucherServiceImpl extends ServiceImpl<VoucherMapper, Voucher> implements IVoucherService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Overridepublic Result queryVoucherOfShop(Long shopId) {// 查询优惠券信息List<Voucher> vouchers = getBaseMapper().queryVoucherOfShop(shopId);// 返回结果return Result.ok(vouchers);}@Override@Transactionalpublic void addSeckillVoucher(Voucher voucher) {// 保存优惠券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());}
}
  1. 基于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.判断库存是否充足 get stockKey
if(tonumber(redis.call('get',stockKey))<=0)then--3.2.库存不足，返回1return 1
end
--3.2.判断用户是否下单SISMEMBER orderKey userId
if(redis.call('sismember',orderKey,userId)== 1)then--3.3.存在，说明是重复下单，返回2return 2
end
--3.4.扣库存 incrby stockKey -1
redis.call("incrby',stockKey,-1)
--3.5.下单(保存用户)sadd orderKey userId
redis.call('sadd',orderkey,userId)
return 0;

  1. 如果抢购成功，将优惠券id和用户id封装后存入阻塞队列

  2. 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

消息队列

1. 认识消息队列

消息队列(Message Queue)，字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色

  • 消息队列:存储和管理消息，也被称为消息代理(Message Broker)
  • 生产者:发送消息到消息队列
  • 消费者:从消息队列获取消息并处理消息
    Redis提供了三种不同的方式来实现消息队列:
  • list结构:基于List结构模拟消息队列
  • PubSub:基本的点对点消息模型
  • Stream:比较完善的消息队列模型
    请添加图片描述

2. 基于List结构模拟消息队列

消息队列(Message Queue)，字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表，很容易模拟出队列效果。
队列是入口和出口不在一边，因此我们可以利用:LPUSH结合 RPOP、或者 RPUSH 结合 LPOP来实现不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回null，并不像IVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。
请添加图片描述
基于List的消息队列有哪些优缺点?
优点:

  • 利用Redis存储，不受限于IVM内存上限
  • 基于Redis的持久化机制，数据安全性有保证
  • 可以满足消息有序性
    缺点:
  • 无法避免消息丢失
  • 只支持单消费者

3. 基于PubSub的消息队列

Pubsub(发布订阅)是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。

  • SUBSCRIBE channel[channel:订阅一个或多个频道
  • PUBLISH channel msg :向一个频道发送消息
  • PSUBSCRlBE pattern[pattern]:订阅与pattern格式匹配的所有频道

通配符：

  • h?llo subscribes to hello, hallo and hxllo
  • h*llo subscribes to hllo and heeeello
  • h[ae]llo subscribes to hello and hallo, but not hillo

在这里插入图片描述
基于Pubsub的消息队列有哪些优缺点?
优点:

  • 采用发布订阅模型，支持多生产、多消费
    缺点:
  • 不支持数据持久化
  • 无法避免消息丢失
  • 消息堆积有上限，超出时数据丢失

4. 基于Stream的消息队列

Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

发送消息的命令:
在这里插入图片描述
例如：
在这里插入图片描述
读取消息的方式之一：XREAD
请添加图片描述
XREAD阻塞方式，读取最新的消息：
在这里插入图片描述
在业务开发中，我们可以循环的调用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下：
在这里插入图片描述
STREAM类型消息队列的XREAD命令特点：

  • 消息可回溯
  • 一个消息可以被多个消费者读取
  • 可以阻塞读取
  • 有消息漏读的风险

5. 基于Stream的消息队列-消费者组

消费者组(Consumer Group):将多个消费者划分到一个组中，监听同一个队列。具备下列特点:

  1. 消息分流
    队列中的消息会分流给组内的不同消费者，而不是重复消费，从而加快消息处理的速度
  2. 消息标示
    消费者组会维护一个标示记录最后一个被处理的消息哪怕消费者宕机重启，还会从标示之后读取消息。确保每一个消息都会被消费
  3. 消息确认
    消费者获取消息后，消息处于pending状态，并存入一个pending-list。当处理完成后需要通过XACK来确认消息，标记消息为已处理，才会从pending-list移除。

创建消费者组:
在这里插入图片描述
key:队列名称
groupName:消费者组名称
ID:起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息MKSTREAM:队列不存在时自动创建队列
其它常见命令:
在这里插入图片描述
从消费者组读取消息:
在这里插入图片描述

  • group:消费组名称
  • consumer:消费者名称，如果消费者不存在，会自动创建一个消费者
  • count:本次查询的最大数量
  • BLOCK milliseconds:当没有消息时最长等待时间
  • NOACK:无需手动ACK，获取到消息后自动确认
  • STREAMS key:指定队列名称
  • ID:获取消息的起始ID:
  • “>”:从下一个未消费的消息开始
  • 其它:根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

消息者监听消息的基本思路：

while(true){// 尝试监听队列，使用阳塞模式，最长等待2000毫秒Object msg = redis.calL("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >")if(msg == null){ // null说明没有消息，继续下一次continue;}try {// 处理消息，完成后一定要ACKhandleMessage(msg);}catch(Exception e){while(true){Object msg= redis.call("XREADGROUP GROUP g1 c1 COUNT 1 STREAMS s1 0");if(msg== null){// null说明没有异常消息，所有消息都已确认，结束循环break;}try {//说明有异常消息，再次处理handleMessage(msg);}catch(Exception e){//再次出现异常，记录日志，继续循环continue;}}}
}

STREAM类型消息队列的XREADGROUP命令特点:

  • 消息可回溯
  • 可以多消费者争抢消息，加快消费速度
  • 可以阻塞读取
  • 没有消息漏读的风险
  • 有消息确认机制，保证消息至少被消费一次

小总结

请添加图片描述

6. 基于Stream消息队列实现异步秒杀

需求:

  1. 创建一个Stream类型的消息队列，名为stream.orders
    请添加图片描述
  2. 修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherld、userld、orderld
--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.库存不足，返回1return 1
end
--3.2.判断用户是否下单SISMEMBER orderKey userId
if(redis.call('sismember',orderKey,userId)== 1)then--3.3.存在，说明是重复下单，返回2return 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;

@Override
public Result seckillVoucher(Long voucherId) {long orderId = redisIdWorker.nextId("order");Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(), voucherId.toString(),UserHolder.getUser().getId().toString(), String.valueOf(orderId));if (result.intValue() != 0) {return Result.fail(result.intValue() == 1 ? "库存不足" : "不能重复下单");}//主线程获取代理对象proxy = (IVoucherOrderService) AopContext.currentProxy();return Result.ok(orderId);
}

String queueName = "stream.orders";private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {while (true) {try {//1. 获取队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),//ReadOffset.lastConsumed()底层就是 '>'StreamOffset.create(queueName, ReadOffset.lastConsumed()));//2. 判断消息是否获取成功if (records == null || records.isEmpty()) {continue;}//3. 消息获取成功之后，我们需要将其转为对象MapRecord<String, Object, Object> record = records.get(0);Map<Object, Object> values = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);//4. 获取成功，执行下单逻辑，将数据保存到数据库中handleVoucherOrder(voucherOrder);//5. 手动ACK，SACK stream.orders g1 idstringRedisTemplate.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 0List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1),StreamOffset.create(queueName, ReadOffset.from("0")));//2. 判断pending-list中是否有未处理消息if (records == null || records.isEmpty()) {//如果没有就说明没有异常消息，直接结束循环break;}//3. 消息获取成功之后，我们需要将其转为对象MapRecord<String, Object, Object> record = records.get(0);Map<Object, Object> values = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);//4. 获取成功，执行下单逻辑，将数据保存到数据库中handleVoucherOrder(voucherOrder);//5. 手动ACK，SACK stream.orders g1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());} catch (Exception e) {log.info("处理pending-list异常");//如果怕异常多次出现，可以在这里休眠一会儿try {Thread.sleep(50);} catch (InterruptedException ex) {throw new RuntimeException(ex);}}}
}
  1. 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单
http://www.dtcms.com/a/347361.html

相关文章：

  • 【JavaEE】多线程 -- JUC常见类和线程安全的集合类
  • 什么猫粮好？2025最新猫粮排名合集
  • 深度解析Bitmap、RoaringBitmap 的原理和区别
  • MySql知识梳理之DDL语句
  • TypeScript 类型系统入门：从概念到实战
  • 从零开始学习JavaWeb-16
  • 阿德莱德多模态大模型导航能力挑战赛！NavBench：多模态大语言模型在具身导航中的能力探索
  • Mysql InnoDB 底层架构设计、功能、原理、源码系列合集【六、架构全景图与最佳实践】
  • 新能源汽车热管理仿真：蒙特卡洛助力神经网络训练
  • android studio配置 build
  • XCVU13P-2FHGB2104E Xilinx（AMD）Virtex UltraScale+ FPGA
  • 力扣热题之多维动态规划
  • [2025CVPR-目标检测方向]学习增量对象检测的内生注意力
  • Redis（18）Redis的键空间通知机制是如何工作的？
  • LangChain4j中集成Redis向量数据库实现Rag
  • 设计模式详解
  • 服务器支持IPv6吗？如何让服务器支持IPv6
  • 疏老师-python训练营-Day54Inception网络及其思考
  • 电阻的标称阻值
  • Python中可以使用中文命名变量、函数、类和方法吗？详细示例与解析
  • Java集合（Collection、Map、转换）
  • JavaScript性能优化实战：从瓶颈识别到极致体验
  • 进阶版蛋白互作研究方法：构建 “体内 + 体外 + 结构 + 功能” 多维度论证体系
  • 场景题：有100个球，其中50个红球和50个黑球，要分配到两个袋子中。然后随机选择一个袋子，再从中随机取一个球，目标是使取到红球的概率最大。
  • n8n 键盘快捷键和控制
  • 数据整理自动化 - 让AI成为你的数据助手
  • Java八股文-java基础面试题
  • 叮小跳APP：自动跳过广告，提升使用体验
  • jQuery 知识点复习总览
  • 在 Spring Boot 中配置和使用多个数据源