网站运营繁忙,招聘网站如何做SEO,如何做企业官网,个人建设网站维护费是多少redis优化秒杀 1. 异步秒杀思路1.1 在redis存入库存和订单信息1.2 具体流程图 2. 实现2.1 总结 3. Redis的消息队列3.1 基于list实现消息队列3.2 基于PubSub实现消息队列3.3 基于stream实现消息队列3.3.1 stream的单消费模式3.3.2 stream的消费者组模式 3.4 基于stream消息队列… redis优化秒杀 1. 异步秒杀思路1.1 在redis存入库存和订单信息1.2 具体流程图 2. 实现2.1 总结 3. Redis的消息队列3.1 基于list实现消息队列3.2 基于PubSub实现消息队列3.3 基于stream实现消息队列3.3.1 stream的单消费模式3.3.2 stream的消费者组模式 3.4 基于stream消息队列实现异步秒杀 本文为学习redis时做的笔记学习内容来自
黑马程序员Redis入门到实战教程该教程是循序渐进的所以不是一上来就讲完最后的解决方案了请耐心看完 1. 异步秒杀思路 这是我们原本的秒杀思路其中的流程都要经过mysql数据库而mysql数据库的并发性能不是很好而且为了避免线程安全问题还加入了分布式锁所以整个流程的性能不好现在我们要去优化它。 我们可以把这整个流程比作一个餐馆点菜的过程前台点菜并将菜品写在小票上给顾客一份后厨一份后厨根据小票的内容依次做菜 根据这个例子我们的流程也可以分为两个部分
第一部分是判断秒杀资格判断秒杀库存和校验一人一单第二部分是减库存创建订单流程
两个部分各自为一个线程主线程判断秒杀资格如果用户有资格就开启一个独立线程完成耗时较久的第二部分 同时我们也要优化判断秒杀资格的性能将库存和订单存入redis中如果判断用户有资格先将优惠券id,用户id、订单id保存在阻塞队列中并将订单id返回给用户用户可以通过这个订单id完成支付操作虽然此时还没有创建订单但是在队列中迟早会创建之后开启独立线程读取队列中的信息完成下单。 现在的业务流程变成了在redis中判断秒杀资格保存信息在队列中并返回订单id,性能和吞吐量大大提高
1.1 在redis存入库存和订单信息
现在讨论这两个东西需要什么样的结构去存储 因为库存只是一个数值我们使用redis中的string类型去存储key是优惠券的id,value是库存 到时只需判断库存是否大于0,如果用户有资格库存要减一相当于在redis中预减库存 因为需要一个优惠券id(key)能存很多用户id而且用户id不能重复所以订单信息我们使用set结构存储 到时只需看value中是否有该用户id来判断该用户是否下过单 1.2 具体流程图 为了保证过程的原子性需要用到lua脚本通过执行lua脚本后的结果来返回异常信息或者订单id这样创建订单的时效性就没有那么强了完全可以照着数据库能承受的范围去执行写的操作用户只需要订单id就能完成支付操作
2. 实现
需求
新增秒杀优惠券的同时将优惠券信息保存在redis中
修改添加秒杀优惠券的方法将优惠券信息保存在redis中
Override
Transactional
public 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());
}添加一个秒杀优惠券
{title: 120元代金券,beginTime: 2023-11-01T01:11:11,actualValue: 12000,shopId: 1,subTitle: 周一至周五均可使用,payValue: 10000,stock: 100,endTime: 2024-11-01T01:11:11,type: 1,rules: 全场通用\\n无需预约\\n可无限叠加\\不兑现、不找零\\n仅限堂食
}基于lua脚本判断秒杀库存、一人一单、决定用户是都抢购成功
写lua脚本
先写参数列表再组合成key最后写逻辑
-- 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. 脚本业务
-- 判断库存是否充足
if (tonumber(redis.call(get,stockKey)) 0) then-- 库存不足return 1
end
-- 判断用户是否下单
if (redis.call(sismember,orderKey,userId) 1) then-- 用户下过单return 2
end-- 扣减库存
redis.call(incrby,stockKey,-1);
-- 下单保存用户id到set集合
redis.call(sadd,orderKey,userId);重写秒杀逻辑
调用lua脚本根据返回的数字判断并返回订单号
private static final DefaultRedisScriptLong SECKILL_SCRIPT;//初始化脚本
static {SECKILL_SCRIPT new DefaultRedisScript();//读取文件位置classpath就是resourceSECKILL_SCRIPT.setLocation(new ClassPathResource(seckill.lua));SECKILL_SCRIPT.setResultType(Long.class);
}Override
public Result seckillVoucher(Long voucherId) {
// 获取用户id
Long userId UserHolder.getUser().getId();
// 1. 调用lua脚本
//不需要传key所以传个空集合
Long result stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(),userId.toString()
);
int intValue result.intValue();//2. 判断结果是否为0
if (intValue ! 0) {return Result.fail(intValue 1 ? 库存不足 : 不要重复下单);
}
//3. 将优惠券id,用户id、订单id保存在阻塞队列中
//TODO 将优惠券id,用户id、订单id保存在阻塞队列中
long orderId redisIdWorker.nextId(order);//4. 返回订单id
return Result.ok(orderId);
}如果抢购成功将优惠券id和用户id封装后存入阻塞队列
//阻塞队列
private BlockingQueueVoucherOrder orderTasks new ArrayBlockingQueue(1024 * 1024);阻塞队列BlockingQueue,当一个线程尝试从一个阻塞队列中获取元素如果队列中没有元素这个线程就会被阻塞当队列中有元素时就会被唤醒并获取元素 seckillVoucher中添加
//3.2 放入阻塞队列中
orderTasks.add(order);开启线程任务不断从阻塞队列中获取信息实现异步下单功能
建立线程池当这个类初始化完线程就去执行VoucherOrderHandler类中的run方法不断获取队列中的订单信息去创建订单 PostConstruct 注解效果是当前类初始化完就去执行 //线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR Executors.newSingleThreadExecutor();
//类加载完就执行
PostConstruct
private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}class VoucherOrderHandler implements Runnable {Overridepublic void run() {while (true) {try {//1.获取队列中的订单信息VoucherOrder voucherOrder orderTasks.take();//2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error(订单异常 e);}}}
}执行handleVoucherOrder方法获取锁最后执行创建订单方法 注1在这个方法中不能使用threadlocal去获取用户信息因为是异步下单这是一个子线程不是主线程没有用户的信息所以从订单中获取用户id 注2与上同理在这个子线程中也无法获取到代理对象将代理对象设置为成员变量再从主线程中获取到代理对象 private void handleVoucherOrder(VoucherOrder voucherOrder) {
Long userId voucherOrder.getUserId();
//创建锁对象
RLock redisLock redissonClient.getLock(lock:order: userId);
//获取锁方法
boolean lockFlag redisLock.tryLock();
//判断是否获取成功
if (!lockFlag) {log.error(不要重复下单);return;
}//事务方法执行起来可能会出现异常但最后都要释放锁,所以try-catch起来
try {proxy.createVoucherOrder(voucherOrder);
} finally {redisLock.unlock();
}
}创建订单逻辑
Transactional
public void createVoucherOrder(VoucherOrder voucherOrder) {
//5.一人一单
Long userId UserHolder.getUser().getId();
//查询
Integer count query().eq(user_id, userId).eq(voucher_id, voucherOrder.getVoucherId()).count();
//判断订单是否存在
if (count 0) {log.error(用户已经购买过一次);return;
}//6.扣减库存
boolean success seckillVoucherService.update().setSql(stock stock-1)
.eq(voucher_id, voucherOrder.getVoucherId())
.gt(stock, 0)
.update();
if (!success) {log.error(不要重复下单);return;
}
save(voucherOrder);
}2.1 总结
整个流程是
主线程
发送请求进入seckillVoucher方法先判断用户是否有秒杀的资格通过lua脚本然后创建订单将用户id、优惠券id、订单id放进订单里将订单放入阻塞队列
子线程
在类初始化的时候去执行线程池线程池的任务是是不断地从队列中获取订单信息然后去创建订单。 创建订单先获取锁再判断一人一单减库存最后执行添加订单方法
完整代码 /*** p* 服务实现类* /p** author 虎哥* since 2021-12-22*/
Slf4j
Service
public class VoucherOrderServiceImpl extends ServiceImplVoucherOrderMapper, VoucherOrder implements IVoucherOrderService {Autowiredprivate ISeckillVoucherService seckillVoucherService;Autowiredprivate RedisIdWorker redisIdWorker;Autowiredprivate StringRedisTemplate stringRedisTemplate;Autowiredprivate RedissonClient redissonClient;private IVoucherOrderService proxy;private static final DefaultRedisScriptLong SECKILL_SCRIPT;//初始化脚本static {SECKILL_SCRIPT new DefaultRedisScript();//读取文件位置classpath就是resourceSECKILL_SCRIPT.setLocation(new ClassPathResource(seckill.lua));SECKILL_SCRIPT.setResultType(Long.class);}//阻塞队列private BlockingQueueVoucherOrder orderTasks new ArrayBlockingQueue(1024 * 1024);//线程池private static final ExecutorService SECKILL_ORDER_EXECUTOR Executors.newSingleThreadExecutor();//类加载完就执行PostConstructprivate void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}class VoucherOrderHandler implements Runnable {Overridepublic void run() {while (true) {try {//1.获取队列中的订单信息VoucherOrder voucherOrder orderTasks.take();//2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error(订单异常 e);}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {Long userId voucherOrder.getUserId();//创建锁对象RLock redisLock redissonClient.getLock(lock:order: userId);//获取锁方法boolean lockFlag redisLock.tryLock();//判断是否获取成功if (!lockFlag) {log.error(不要重复下单);return;}//事务方法执行起来可能会出现异常但最后都要释放锁,所以try-catch起来try {proxy.createVoucherOrder(voucherOrder);} finally {redisLock.unlock();}}Overridepublic Result seckillVoucher(Long voucherId) {// 获取用户idLong userId UserHolder.getUser().getId();// 1. 调用lua脚本//不需要传key所以传个空集合Long result stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());int intValue result.intValue();//2. 判断结果是否为0if (intValue ! 0) {return Result.fail(intValue 1 ? 库存不足 : 不要重复下单);}//3. 将优惠券id,用户id、订单id保存在阻塞队列中//3.1.创建订单VoucherOrder order new VoucherOrder();//3.1.1 订单idlong orderId redisIdWorker.nextId(order);order.setId(orderId);//3.1.2 用户idorder.setUserId(userId);//3.1.3 优惠券idorder.setVoucherId(voucherId);//3.2 放入阻塞队列中orderTasks.add(order);//3.3 获取当前的代理对象(事物)proxy (IVoucherOrderService) AopContext.currentProxy();//4. 返回订单idreturn Result.ok(orderId);}Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {//5.一人一单Long userId voucherOrder.getUserId();//查询Integer count query().eq(user_id, userId).eq(voucher_id, voucherOrder.getVoucherId()).count();//判断订单是否存在if (count 0) {log.error(用户已经购买过一次);return;}//6.扣减库存boolean success seckillVoucherService.update().setSql(stock stock-1).eq(voucher_id, voucherOrder.getVoucherId()).gt(stock, 0).update();if (!success) {log.error(不要重复下单);return;}save(voucherOrder);}}测试 现在我们的阻塞队列使用的是jvm的内存将来有无数的请求进来内存可能会满而且如果服务重启或者宕机订单信息就消失了可能会出现一些问题接下来学习一下redis的消息队列
3. Redis的消息队列
消息队列字面意思就是存放消息的队列最简单的消息队列包含三个角色
生产者发送消息到消息队列消息队列存储和管理消息也称为消息代理消费者从消息队列获取消息并处理消息 市面上有很多的消息队列的产品但是搭建他们也是需要成本的既然我们已经搭建起了redis集群为了减少成本可以使用redis提供的三种不同的方式
list结构基于list结构模拟消息队列PubSub订阅发布基本的点对点消息模型Stream比较完善的消息队列模型
3.1 基于list实现消息队列
Redis的list数据结构是一个双向链表很容易模拟出队列效果。 队列是入口和出口不在一边因此我们可以利用: LPUSH 结合 RPOP、或者 RPUSH 结合LPOP来实现。 不过要注意的是当队列中没有消息时RPOP或LPOP操作会返回null并不像VM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。
优点:
利用Redis存储不受限于JVM内存上限基于Redis的持久化机制数据安全性有保证可以满足消息有序性
缺点:
无法避免消息丢失只支持单消费者
3.2 基于PubSub实现消息队列
Pubsub(发布订阅)是Redis2.0版本引入的消息传递模型。顾名思义消费者可以订阅一个或多个channel生产者向对应channel发送消息后所有订阅者都能收到相关消息
SUBSCRIBE channel[channell : 订阅一个或多个频道PUBLISH channel msg : 向一个频道发送消息PSUBSCRIBE pattern[pattern]: 订与pattern格式匹配的所有频道 优点:
采用发布订阅模型支持多生产、多消费
缺点:
不支持数据持久化无法避免消息丢失消息堆积有上限超出时数据丢失
3.3 基于stream实现消息队列
3.3.1 stream的单消费模式
Stream是Redis5.0引入的一种新数据类型可以实现一个功能非常完善的消息队列
发送消息的命令 中间的内容都是可选。红框标注的基本可以不用管绿框用来设置消息队列的最大消息数量黄框用来设置消息的id,*代表redis自动生成蓝框是队列中的消息内容
示例 xadd users * name zhuyi love lvhan 返回id
查看队列中消息数量的命令
xlen key 示例
读取队列中消息的命令 [COUNT count] 是每次读取消息的最大数量[BLOCK milliseconds] 当没消息时是否阻塞及阻塞时长STREAMS key 要从哪个队列开始读key就是队列名ID 起始id,只返回大于该id的消息。0代表从第一个消息开始$代表从最新的消息开始
示例 从第一个消息开始读 因为消息已经读过没有最新的消息所以读不出来
特点
消息可回溯一个消息可以被多个消费者读取可以阻塞读取有消息漏读的风险。使用$在读一条消息的时候有超过一条以上的消息进入队列只会读取最后一条
3.3.2 stream的消费者组模式
消费者组将多个消费者划分到一个组中监听同一个队列有以下好处
消息分流队列中的消息会分流给组内的不同消费者而不是重复消费从而加快消息处理的速度。如果想让一个消息被多个消费者消费可以多加几个组消息标识消费者组会维护一个标识记录最后一个被处理的消息哪怕消费者宕机重启还会从标识之后读取消息确保每一个消息都会被消费消息确认消费者获取消息后消息处于pending待处理状态并存入一个pending-list。当处理完成后需要通过XACK来确认消息标记消息为已处理才会从pending-list移除保证所有的消息只要被获取到了就能至少被消费一次
创建消费者组 key队列名称groupName消费者组名称ID起始ID标识$代表队列中最新的消息0则代表队列中第一个消息MKSTREAM队列不存在时自动创建队列
其他常见命令
示例
从消费者组读取消息 group消费组名称consumer消费者名称如果消费者不存在会自动创建一个消费者count本次查询的最大数量BLOCK millisecond当没有消息时最长等待时间NOACK无需手动ACK获取到消息后自动确认STREAMS key指定队列名称ID获取消息的起始id1. “”从下一个未消费的消息开始。2. 其他根据指定id从pending-list中获取已消费但未确认的消息例如0是从pending-list中的第一个消息开始
示例
确认消息
我们获取到消息消费后一定要确认它把他从pending-list中移除
key队列名称group消费组名称ID获取消息的起始id1. “”从下一个未消费的消息开始。2. 其他根据指定id从pending-list中获取已消费但未确认的消息例如0是从pending-list中的第一个消息开始
示例
消费者监听消息的基本思路 特点
消息可回溯可以多消费者争抢消息加快消费速度可以阻塞读取没有消息漏读的风险有消息确认机制保证消息至少被消费一次
如果你的公司业务比较庞大对消息队列要求比较严格还是建议使用更专业的消息队列如rabbitmq等但如果是中小型公司对消息队列需要没那么大redis的stream就已经能满足需求了
3.4 基于stream消息队列实现异步秒杀 创建一个Stream类型的消息队列定义为stream.orders,这里直接在客户端完成了
XGROUP CREATE stream.orders g1 0 mkstream修改之前的秒杀下单lua脚本在认定有抢购资格后直接向stream.orders中添加消息内容包括voucherId、userId、orderId
修改lua脚本 主要添加了一个订单id的参数在业务的最后向队列发送消息
-- 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. 脚本业务
-- 判断库存是否充足
if (tonumber(redis.call(get,stockKey)) 0) then-- 库存不足return 1
end
-- 判断用户是否下单
if (redis.call(sismember,orderKey,userId) 1) then-- 用户下过单return 2
end-- 扣减库存
redis.call(incrby,stockKey,-1);
-- 下单保存用户id到set集合
redis.call(sadd,orderKey,userId);
-- 发送消息
redis.call(xadd,stream.orders,*,voucherId,voucherId,userId,userId,id,orderId)
return 0修改一下调用lua脚本的逻辑 新添加一个订单id的参数
// 1. 调用lua脚本
//不需要传key所以传个空集合
Long result stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(),String.valueOf(orderId)
);
int intValue result.intValue();项目启动时开启一个线程任务尝试获取stream.orders中的消息完成下单
在原有的线程任务逻辑上修改从消息队列中获取订单信息判断一下订单信息是否为空如果为空说明没有消息继续下一次循环如果有去解析数据拿到订单通过以前写过的createVoucherOrder()方法来创建订单最后一定要确认消息将消息从pending-list中移除。 如果在执行时出现了错误或者服务宕机通过handlePendingList()方法处理pending-list中已消费但未确认的订单这里如果出现异常就不用再调用这个方法了
private class VoucherOrderHandler implements Runnable {Overridepublic void run() {while (true) {try {// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 ListMapRecordString, Object, Object list stringRedisTemplate.opsForStream().read(Consumer.from(g1, c1),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(stream.orders, ReadOffset.lastConsumed()));// 2.判断订单信息是否为空if (list null || list.isEmpty()) {// 如果为null说明没有消息继续下一次循环continue;}// 解析数据MapRecordString, Object, Object record list.get(0);MapObject, Object value record.getValue();VoucherOrder voucherOrder BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单handleVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge(stream.orders, 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 s1 0ListMapRecordString, Object, Object list stringRedisTemplate.opsForStream().read(Consumer.from(g1, c1),StreamReadOptions.empty().count(1),StreamOffset.create(stream.orders, ReadOffset.from(0)));// 2.判断订单信息是否为空if (list null || list.isEmpty()) {// 如果为null说明没有异常消息结束循环break;}// 解析数据MapRecordString, Object, Object record list.get(0);MapObject, Object value record.getValue();VoucherOrder voucherOrder BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单handleVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge(stream.orders, g1, record.getId());} catch (Exception e) {log.error(处理订单异常, e);}}}
}XACK确认消息 stringRedisTemplate.opsForStream().acknowledge(s1, g1, record.getId());
