Redis之优惠券秒杀

news2024/11/20 9:12:54

文章目录

  • 全局ID生成器
  • 添加优惠券
  • 实现优惠券秒杀下单
  • 超卖问题
    • 悲观锁和乐观锁相关文章
    • 乐观锁执行逻辑
    • 乐观锁解决超卖问题
  • 一人一单功能
    • 超卖问题相关文章
    • 一人一单执行逻辑
    • 代码实现
    • 集群模式下锁失效
  • 分布式锁
    • 基于Redis的分布式锁
    • Redis实现分布式锁流程
    • 实现分布式锁初级版本
    • 分布式锁误删问题
    • 改进分布式锁防止误删问题
    • 分布式锁的原子性问题
    • Lua脚本解决多条命令原子性问题
    • 分布式锁总结
  • 分布式锁优化-Redisson
    • 快速入门
    • Redission可重入锁原理
    • Redisson分布式锁原理
    • Redisson分布式锁总结
    • Redisson的MultiLock原理
    • 锁重试和WatchDog机制和MutiLock原理
  • 分布式锁总结
  • Redis优化秒杀
    • 秒杀业务的优化思路是什么?
    • 基于阻塞队列的异步秒杀存在哪些问题?
  • Redis消息队列实现异步秒杀
    • 消息队列的定义
    • 基于List结构模拟消息队列
    • 基于PubSub的消息队列
    • 基于Stream的消息队列
    • 基于Stream的消息队列-消费者组
      • 创建消费者组和其他常用命令
      • 从消费者读取消息
    • 基于消费者组实现消息监听的基本思路
    • Redis消息队列总结
  • 基于Redis的Stream消息队列, 实现异步秒杀下单

全局ID生成器

特性:

  • 唯一性
  • 高可用
  • 高性能
  • 递增性
  • 安全性

image.png

全局唯一ID生成策略:

  • UUID
  • Redis自增
  • snowflake ( 雪花算法 )
  • 数据库自增

Redis自增ID策略:

  • 每天一个Key, 方便统计订单量
  • ID构造 -> 时间戳 + 计数器
/** 开始时间截 (2022-01-01) */
private static final long BEGIN_TIMESTAMP = 1640995200L;
/** 序列号位数 */
private static final int COUNT_BITS = 32;

private StringRedisTemplate stringRedisTemplate;
public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
	this.stringRedisTemplate = stringRedisTemplate;
}

public long nextId(String keyPrefix){
	// 1. 生成时间戳
	LocalDateTime now = LocalDateTime.now();
	long nowSecond = now.toEpochSecond(ZoneOffset.UTC); // 当前秒数
	long timeStamp = nowSecond - BEGIN_TIMESTAMP; // 时间戳
	// 2. 生成序列号
	// 2.1 获取当前日期 ( 精确到天 )
	String date = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
	// 2.2 自增长
	long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
	// 3. 拼接并返回
	return timeStamp << COUNT_BITS | count;
}
  1. 拿到当前秒数时间戳nowSecond与开始时间戳BEGIN_TIMESTAMP进行相减, 得时间戳timeStamp
  2. 生成序列号, 序列号由key和当前时间date来进行拼接由Redis自增生成
  3. 最后用位运算来将时间戳和序列号进行拼接, 因为最后得到的是数值, 而不是字符串, 不能直接拼接

添加优惠券

/**
 * 新增秒杀券
 * @param voucher 优惠券信息，包含秒杀信息
 * @return 优惠券id
 */
@PostMapping("seckill")
public Result addSeckillVoucher(@RequestBody Voucher voucher) {
    voucherService.addSeckillVoucher(voucher);
    return Result.ok(voucher.getId());
}

@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);
}
  1. 阿巴阿巴阿巴~ 太简单了 自行理解 哈哈哈~

实现优惠券秒杀下单

image.png

/**
 * 秒杀优惠券下单
 * @param voucherId 优惠券id
 * @return 订单id
 */
@PostMapping("seckill/{id}")
public Result seckillVoucher(@PathVariable("id") Long voucherId) {
    return voucherOrderService.seckillVoucher(voucherId);
}

@Override
@Transactional
public Result seckillVoucher(Long voucherId) {
	// 1.查询优惠券
	SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
	// 2.判断秒杀是否开始
	if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
		return Result.fail("秒杀还未开始");
	}
	// 3.判断秒杀是否结束
	if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
		return Result.fail("秒杀已经结束");
	}
	// 4.判断库存是否充足
	if (voucher.getStock() < 1) {
		return Result.fail("库存不足");
	}
	// 5.扣减库存
	boolean result = lambdaUpdate()
			.setSql("stock = stock - 1")
			.eq(VoucherOrder::getVoucherId, voucherId).update();
	if(!result){
		return Result.fail("库存不足");
	}
	// 6.创建订单
	VoucherOrder order = new VoucherOrder();
	// 6.1.订单ID
	long orderId = redisIdWorker.nextId("order");
	order.setId(orderId);
	// 6.2.用户ID
	Long userId = UserHolder.getUser().getId();
	order.setUserId(userId);
	// 6.3.代金券ID
	order.setVoucherId(voucherId);
	save(order);
	// 7.返回订单ID
	return Result.ok(orderId);
}
  1. 查询优惠券->判断秒杀时间->判断库存->下单->更新库存
  2. 判断秒杀时间要判断起始时间和结束时间, 看是否在时间区间内
  3. 判断库存是否充足, 充足则进行库存更新, 使用lambdaUpdate来更新
  4. 创建订单, 更新库存, 使用之前的全局唯一ID生成器作为订单ID
  5. 在设置用户ID, 秒杀券ID的属性, 最后保存秒杀券信息

超卖问题

悲观锁和乐观锁相关文章

乐观锁和悲观锁的优缺点与适用场景-CSDN博客

乐观锁执行逻辑

image.png

  1. 如果修改时的库存与前面查询到的库存一致则进行更改
  2. 不进行更改, 但一时间涌入大量的请求到数据库, 查到的数据很有可能是相同的
  3. 所以修改的条件并不是修改前后的库存一致, 而是修改时库存 > 0即可
  4. 参考下面的示例代码

乐观锁解决超卖问题

// 5.扣减库存
boolean result = seckillVoucherService.lambdaUpdate()
		.setSql("stock = stock - 1")
		.eq(SeckillVoucher::getVoucherId, voucherId) // where id = ? and stock > 0 
		.gt(SeckillVoucher::getStock, 0) // 乐观锁 在原来代码的基础上做"修修"的改动即可
		.update();
if(!result){
	return Result.fail("库存不足");
}

小结一下:
悲观锁: 添加同步锁, 让线程串行执行

  • 优点: 简单粗暴
  • 缺点: 性能一般

乐观锁: 不加锁, 在更新时判断是否有其他线程在修改

  • 优点: 性能好
  • 缺点: 存在成功率低的问题

一人一单功能

超卖问题相关文章

【Redis】电商项目秒杀问题之超卖问题与一人一单问题_redis超卖_西瓜霜润喉片的博客-CSDN博客

一人一单执行逻辑

image.png

代码实现

/**
 * 秒杀优惠券下单
 * @param voucherId
 * @return
 */
@Override
public Result seckillVoucher(Long voucherId) {
	// 1.查询优惠券
	SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
	// 2.判断秒杀是否开始
	if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
		return Result.fail("秒杀还未开始");
	}
	// 3.判断秒杀是否结束
	if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
		return Result.fail("秒杀已经结束");
	}
	// 4.判断库存是否充足
	if (voucher.getStock() < 1) {
		return Result.fail("库存不足");
	}
	// 5.扣减库存
	boolean result = seckillVoucherService.lambdaUpdate()
			.setSql("stock = stock - 1")
			.eq(SeckillVoucher::getVoucherId, voucherId)
			.gt(SeckillVoucher::getStock, 0) // 乐观锁
			.update();
	if(!result){
		return Result.fail("库存不足");
	}
	Long userId = UserHolder.getUser().getId();
	synchronized (userId.toString().intern()) {
		// 获取代理对象(事务)
		IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
		return proxy.createVoucherOrder(voucherId);
	}
}

@Transactional
public Result createVoucherOrder(Long voucherId) {
	// 6.一人一单
	Long userId = UserHolder.getUser().getId();
	// 6.1.查询订单
	Long count = lambdaQuery()
			.eq(VoucherOrder::getUserId, userId)
			.eq(VoucherOrder::getVoucherId, voucherId)
			.count();
	// 6.2.判断是否存在
	if(count > 0){
		return Result.fail("每人限购一张");
	}
	// 7.创建订单
	VoucherOrder order = new VoucherOrder();
	// 7.1.订单ID
	long orderId = redisIdWorker.nextId("order");
	order.setId(orderId);
	// 7.2.用户ID
	order.setUserId(userId);
	// 7.3.代金券ID
	order.setVoucherId(voucherId);
	save(order);
	// 8.返回订单ID
	return Result.ok(orderId);
}
  1. 一人一单防止超卖问题, 使用悲观锁来限制优惠券的购买
  2. 通过悲观锁确保了同一用户创建订单的并发安全性，而使用事务则保证了创建订单的原子性和一致性
  3. 在秒杀时间内且库存充足情况下, 会去查询当前用户ID是否已购买当前秒杀券
  4. 但短时间大量请求打到数据库上, 查出来的结果部分可能相同, 这依然存在超卖问题
  5. 这时单独将查询是否已购买秒杀券和创建订单加上悲观锁, 但此时又存在性能低下
  6. 每个线程都要去竞争同一个锁, 出现响应慢的情况, 只用对每个用户加锁即可, 而不是线程加锁
  7. synchronized (userId.toString().intern())
  • 这是为了保证在多线程环境下，同一用户的订单创建操作不会发生并发冲突
  • 通过 synchronized 关键字，使用用户ID的字符串形式作为锁对象进行同步
  • 通过调用 toString() 将 userId 转换成字符串, intern() 方法获取字符串的常量池对象，以确保锁的唯一性
  1. IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy()
  • 这是用来解决前面的事务还没提交而产生的并发问题使事务失效, 导致超卖问题
  • 通过 AopContext.currentProxy() 获取当前代理对象，用于在事务中调用 createVoucherOrder 方法
  • 这样可以确保事务生效，因为在同一类内调用带有 @Transactional 注解的方法时，默认不会触发事务

集群模式下锁失效

  1. 现在, 用户请求会在这两个节点上负载均衡, 再次测试下发现存在线程安全问题。
  2. 但是在集群模式下，加锁只是该台jvm给当前这台服务器处理的请求加锁, 而集群是多台服务器轮询处理请求，会造成每台服务器都有一个加锁的线程, 每台服务器都会有一个新订单创建处理

image.png

分布式锁

分布式锁:
满足分布式系统或集群模式下多进程可见并且互斥的锁

image.png

image.png

基于Redis的分布式锁

实现分布式锁时需要实现的两个基本方法:

  • 获取锁:

  • 互斥 确保只有一个线程获取锁

  • SET lock thread1 NX EX 10 // 添加锁 NX是互斥 EX是设置超时时间

  • 释放锁:

  • 手动释放

  • DEL key // 释放锁 删除即可

Redis实现分布式锁流程

image.png

实现分布式锁初级版本

@Override
public boolean tryLock(long timeoutSec) {
	// 获取线程标识
	long threadId = Thread.currentThread().getId();
	// 获取锁
	Boolean success = stringRedisTemplate.opsForValue()
			.setIfAbsent(KEY_PREFIX + name, String.valueOf(threadId), timeoutSec, TimeUnit.SECONDS);
	return Boolean.TRUE.equals(success); // 防止拆箱时出现空指针
}
@Override
public void unlock() {
	// 释放锁
	stringRedisTemplate.delete(KEY_PREFIX + name);
}
  1. 获取锁 SET key name NX EX 10 lock:name->key, threadId->name
  2. Boolean.TRUE.equals(success)防止拆箱的时候出现NPE
  3. 释放锁直接删除锁就行~

分布式锁误删问题

问题现象
image.png

解决方案
image.png

改进分布式锁防止误删问题

image.png

  1. 在获取锁时存入线程标识, ( 可以使用UUID来表示 )
  2. 在释放锁时先获取锁中的线程标识, 判断是否与当前线程标识一致
  • 如果一致则释放锁
  • 如果不一致则不释放锁
private static final String KEY_PREFIX = "lock:";
private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
@Override
public boolean tryLock(long timeoutSec) {
	// 获取线程标识
	String threadId = ID_PREFIX + Thread.currentThread().getId();
	// 获取锁
	Boolean success = stringRedisTemplate.opsForValue()
			.setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
	return Boolean.TRUE.equals(success); // 防止拆箱时出现空指针
}
@Override
public void unlock() {
	// 获取线程标识
	String threadId = ID_PREFIX + Thread.currentThread().getId();
	// 获取锁中的标识
	String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
	if(!threadId.equals(id)) {
		// 释放锁
		stringRedisTemplate.delete(KEY_PREFIX + name);
	}
}
  1. 原来只有线程ID作为标识, 当在集群的时候, 会有可能出现线程ID一致的情况, 此时就有可能锁误释放问题
  2. 要在线程ID前面拼接上一段随机UUID标识作为前缀, 大大增加的线程的随机性, 减少ID碰撞的可能性
  3. 获取锁的时候线程标识加UUID, 释放锁的时候要进行线程标识和锁标识的比较, 如果标识相同才可以释放

分布式锁的原子性问题

image.png

Lua脚本解决多条命令原子性问题

image.png

image.png

在resource目录下创建一个lua脚本
image.png

if(redis.call('get', KEYS[1]) == ARGV[1]) then
    -- 释放锁 del key
    return redis.call('del', KEYS[1])
end
return 0

private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
static {
	UNLOCK_SCRIPT = new DefaultRedisScript<>();
	UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
	UNLOCK_SCRIPT.setResultType(Long.class);
  1. 初始化Lua脚本, 使用静态代码块, 找到resource目录下的unlock.lua脚本
@Override
public void unlock() {
	// 调用lua脚本
	stringRedisTemplate.execute(
			UNLOCK_SCRIPT,
			Collections.singletonList(KEY_PREFIX + name),
			ID_PREFIX + Thread.currentThread().getId());
}
  1. 最后调用Lua脚本, 使获取线程标识和释放锁合并执行, 相当于事务一样, 增强原子性

分布式锁总结

  1. 基于Redis分布式锁实现思路
  • 利用set nx ex获取锁, 并设置过期时间, 保存线程标识
  • 释放锁时先判断线程标识是否与自己一致, 一致则删除锁
  1. 特性
  • 利用set nx满足互斥性
  • 利用set ex保证故障时锁依旧能释放, 避免死锁, 提高安全性
  • 利用Redis集群保证高可用和高并发特性

分布式锁优化-Redisson

  1. 官网地址: https://redisson.org
  2. GitHub地址: https://github.com/redisson/redisson

快速入门

  1. 引入依赖
<!--Redisson-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.25.0</version>
</dependency>
  1. 创建Redission客户端RedisClient
@Configuration
public class RedissonConfig {
	@Bean
	public RedissonClient redissonClient(){
		// 配置
		Config config = new Config();
		config.useSingleServer().setAddress("redis://192.168.1.112:6379");
		// 创建RedissonClient对象
		return Redisson.create(config);
	}
}
  1. 注入RedisClient客户端, 并使用
@Resource
private RedissonClient redissonClient;

// 创建锁对象
RLock lock = redissonClient.getLock("lock:order:" + userId);
// 获取锁
boolean isLock = lock.tryLock();
// 判断是否获取锁成功
if(!isLock){
	return Result.fail("不允许重复下单");
}

Redission可重入锁原理

  1. 获取锁Lua脚本

image.png

  1. 释放锁Lua脚本

image.png

Redisson分布式锁原理

image.png

Redisson分布式锁总结

  1. 可重入: 利用Hash结构记录线程ID和重入次数
  2. **可重试: **利用信号量和PubSub功能实现等待, 唤醒, 获取锁失败的重试机制
  3. **超时续约: **利用WatchDog, 每隔一段时间( releaseTime / 3 ) 重置超时时间

看门狗机制：
如果是没有传入时间，则此时也会进行抢锁， 而且抢锁时间是默认看门狗时间30s，
ttlRemainingFuture.onComplete((ttlRemaining, e) 这句话相当于对以上抢锁进行了监听，也就是说当上边抢锁完毕后，此方法会被调用，具体调用的逻辑就是去后台开启一个线程，进行续约逻辑，也就看门狗线程。
看门狗线程会开启一个定时任务：每10s会重置锁的有效期，时间一到，这个定时任务就触发了，它就会去续约，把当前这把锁续约成30s，如果操作成功，那么此时就会递归调用自己，在重新设置一个定时任务，于是在过10s又会去续约，完成不停的续约，这样锁就不会过期了。
直到什么时候这个定时任务会停止呢，当这把锁被完全释放的时候就会被删除。或者服务宕机了。

Redisson的MultiLock原理

主从一致性问题：
Redis提够了主从集群，主从同步存在延迟，当主服务宕机时，如果从并没有同步主中的数据，则会出现锁失效。
如果我们只有一个Redis主机，那么如果这个Redis主机发生了故障，那么所以需要Redis服务的都会发生问题，也包括分布式锁。
所以为了解决单主机的问题，我们需要搭建一个Redis集群。主从集群就是有一些主机器用来做写操作，从机器用来做读操作，主机器一旦有命令进来，那么从机器就会去同步主机器上的数据，来保证主从一致性问题。
假设我们现在有三个机器，一个是主，两个从。此时我们去写命令，写在主机上，主机会将数据同步给从机，但是假设还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并让一个从变成主，而此时新的主实际上并没有锁信息，此时锁信息就已经丢掉了。

为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的， 这把锁 加锁的逻辑需要写入到每一个主从节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。因为有的节点上已经有锁了，有的因为挂了从节点就拿不到锁就不能够成功。

当我们去设置了多个锁时，redission会将多个锁添加到一个集合中，然后用while循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是4500ms，假设在这4500ms内，所有的锁都加锁成功， 那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试.

锁重试和WatchDog机制和MutiLock原理

这里有相关源码的解析, 可以看看
【Redis】分布式锁的应用以及Redission看门狗机制和MultiLock的源码深入解析_redission分布式锁看门狗-CSDN博客

分布式锁总结

  1. 不可重入Redis分布式锁
  • 原理: 利用setnx的互斥性, 利用ex避免死锁, 释放锁时判断线程标识
  • 缺陷: 不可重入, 无法重试, 锁超时失效
  1. 可重入的Redis分布式锁
  • 原理: 利用Hash结构, 记录线程标识和重入次数, WatchDog延续锁时间, 信号量控制锁重试等待
  • 缺陷: redis宕机引起锁失效问题
  1. Redisson的MutiLock
  • 原理: 多个独立的Redis节点, 必须在所有节点都获取重入锁才算获取锁成功
  • 缺陷: 运维成本高, 实现复杂

Redis优化秒杀

优化思路
image.png

  1. 新增秒杀优惠券的同时, 将优惠券信息保存到Redis中

  2. 基于Lua脚本, 判断秒杀库存, 一人一单, 决定用户是否抢购成功

  3. 如果抢购成功, 将优惠券ID和用户ID封装后存入阻塞队列

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

  5. 保存优惠券的同时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());
}
  1. seckill.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.1.1.库存不足，返回1
    return 1
end
-- 3.2.判断用户是否已下单 sismember orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
    -- 3.2.1.已下单，返回2
    return 2
end
-- 3.3.减库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.4.下单(保存用户) sadd orderKey userId
redis.call('sadd', orderKey, userId)
return 0

private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
	SECKILL_SCRIPT = new DefaultRedisScript<>();
	SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
	SECKILL_SCRIPT.setResultType(Long.class);
}

/**
 * 秒杀优惠券下单
 * @param voucherId
 * @return
 */
private IVoucherOrderService proxy;
@Override
public Result seckillVoucher(Long voucherId) {
	// 获取用户
	Long userId = UserHolder.getUser().getId();
	// 1. 执行lua脚本
	Long result = stringRedisTemplate.execute(
			SECKILL_SCRIPT,
			Collections.emptyList(),
			voucherId.toString(), userId.toString()
	);
	// 2. 判断结果是否为0
	int r = result.intValue();
	if(r != 0){
		// 2.1.不为0, 代表没有购买资格
		return Result.fail(r == 1 ? "库存不足" : "不可重复下单");
	}
	// 2.2.为0, 有购买资格, 把下单信息保存到阻塞队列
	VoucherOrder voucherOrder = new VoucherOrder();
	// 2.3.订单ID
	long orderId = redisIdWorker.nextId("order");
	voucherOrder.setId(orderId);
	// 2.4.用户ID
	voucherOrder.setUserId(userId);
	// 2.5.代金券ID
	voucherOrder.setVoucherId(voucherId);
	// 2.6.放入阻塞队列
	orderTasks.add(voucherOrder);
	// 3.获取代理对象
	proxy = (IVoucherOrderService)AopContext.currentProxy();
	// 3. 返回订单ID
	return Result.ok(orderId);
}
  1. 初始化seckill.lua脚本, 修改秒杀下单逻辑
  2. Lua脚本实现用Redis对订单库存判断和一人一单判断
  3. 有购买资格就将订单信息和用户信息存入阻塞队列实行异步下单
/**
 * 阻塞队列
 */
private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
/**
 * 处理订单的线程池
 */
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

/**
 * 初始化
 */
@PostConstruct
private void init(){
	SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}

/**
 * 处理订单的线程
 */
private class VoucherOrderHandler implements Runnable{
	@Override
	public void run(){
		while(true){
			try{
				// 1.获取队列中的订单信息
				VoucherOrder voucherOrder = orderTasks.take();
				// 2.创建订单
				handleVoucherOrder(voucherOrder);
			}catch(Exception e){
				log.error("处理订单异常", e);
			}
		}
	}
}

/**
 * 处理订单
 * @param voucherOrder
 */
private void handleVoucherOrder(VoucherOrder voucherOrder) {
	// 1.获取用户
	Long userId = voucherOrder.getUserId();
	// 2.创建锁对象
	RLock lock = redissonClient.getLock("lock:order:" + userId);
	// 3.获取锁
	boolean isLock = lock.tryLock();
	// 4.判断是否获取锁成功
	if(!isLock){
		// 获取锁失败, 返回错误或重试
		log.error("不允许重复下单");
		return;
	}
	try{
		proxy.createVoucherOrder(voucherOrder);
	}finally{
		// 释放锁
		lock.unlock();
	}
}

/**
 * 创建订单
 * @param voucherOrder
 */
@Transactional
public void createVoucherOrder(VoucherOrder voucherOrder) {
	// 6.一人一单
	Long userId = voucherOrder.getUserId();
	// 6.1.查询订单
	Long count = lambdaQuery()
			.eq(VoucherOrder::getUserId, userId)
			.eq(VoucherOrder::getVoucherId, voucherOrder.getId())
			.count();
	// 6.2.判断是否存在
	if(count > 0){
		log.error("每人限购一张");
		return;
	}
	boolean success = seckillVoucherService.update()
			.setSql("stock = stock - 1")// set stock = stock - 1
			.eq("voucher_id", voucherOrder)
			.gt("stock", 0) // where id = ? and stock > 0
			.update();
	if(!success){
		// 扣减失败
		log.error("库存不足");
		return;
	}
	// 7.创建订单
	save(voucherOrder);
}
  1. 首先会创建一个阻塞队列和处理订单的线程池
  2. 用一个while循环处理阻塞队列里的订单, 处理订单要先拿到锁
  3. 最后创建订单信息, 判断数据库是否已存在订单信息, 防止重复下单
  4. 最后更新数据库订单库存和订单信息

秒杀业务的优化思路是什么?

  • 先利用Redis完成库存余量, 一人一单判断, 完成抢单业务
  • 再将下单业务放入阻塞队列, 利用独立线程异步下单

基于阻塞队列的异步秒杀存在哪些问题?

  • 内存限制问题
  • 数据安全问题

Redis消息队列实现异步秒杀

消息队列的定义

image.png

基于List结构模拟消息队列

  1. Redis 的 list 数据结构是一个双向链表, 很容易模拟出队列效果
  2. 队列是入口和出口不在一边, 因此我们可以利用: LPUSH结合RPOP, 或者RPUSH和LPOP
  3. 当队列中没有消息时RPOP或LPOP操作会返回NULL, 并不像JVM的阻塞队列会一直等待消息
  4. 因此这里应该使用BRPOP或者BLPOP来实现阻塞效果

image.png
优点:

  • 利用Redis’存储, 不受限于JVM内存上限
  • 基于Redis持久化机制, 数据安全性有保证
  • 可以满足消息有序性

缺点:

  • 无法避免消息丢失
  • 只支持单消费者

基于PubSub的消息队列

image.png
优点:

  • 采用发布订阅模型, 支持多生产, 多消费

缺点:

  • 不支持数据持久化
  • 无法避免消息丢失
  • 消息堆积有上限, 超出时数据丢失

基于Stream的消息队列

发送消息
image.png

读取消息
image.png

实现持续监听队列
image.png

:::danger
但是可能会出现消息漏读的情况
:::
image.png

Stream类型消息队列的XREAD命令特点:

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

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

image.png

  1. 消息确认正好可以弥补有消息漏读的风险~

创建消费者组和其他常用命令

image.png

从消费者读取消息

image.png

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

基于消费者组实现消息监听的基本思路

image.png

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

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

Redis消息队列总结

image.png

基于Redis的Stream消息队列, 实现异步秒杀下单

  1. 创建一个Stream类型的消息队列, 名为stream.order
  2. 修改之前的秒杀下单Lua脚本, 在认定有抢购资格之后, 直接向stream.order中添加消息

内容包括voucherId, userId, orderId

  1. 项目启动时, 开启一个线程任务, 尝试获取stream.order中的消息, 完成下单
/**
 * 秒杀优惠券下单
 *
 * @param voucherId
 * @return
 */
private IVoucherOrderService proxy;
@Override
public Result seckillVoucher(Long voucherId) {
	// 获取用户
	Long userId = UserHolder.getUser().getId();
	// 获取订单ID
	long orderId = redisIdWorker.nextId("order");
	// 1. 执行lua脚本
	Long result = stringRedisTemplate.execute(
			SECKILL_SCRIPT,
			Collections.emptyList(),
			voucherId.toString(), userId.toString(), String.valueOf(orderId)
	);
	// 2. 判断结果是否为0
	int r = result.intValue();
	if (r != 0) {
		// 2.1.不为0, 代表没有购买资格
		return Result.fail(r == 1 ? "库存不足" : "不可重复下单");
	}
	// 3.获取代理对象
	proxy = (IVoucherOrderService) AopContext.currentProxy();
	// 3. 返回订单ID
	return Result.ok(orderId);
}

/** Lua下单脚本 */
-- 3.3.减库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.4.下单(保存用户) 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
  1. 判断库存和下单资格, 如果是1返回库存不足, 是2则是无下单资格
  2. 返回0则有下单资格, Redis减库存和保存下单记录, 最后将订单消息保存到消息队列中
  3. 将订单id返回给用户
/**
 * 处理订单的线程
 */
private class VoucherOrderHandler implements Runnable {
	String queueName = "streams.order";
	@Override
	public void run() {
		while (true) {
			try {
				// 1.获取队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS streams.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.没有消息, 休眠一会
					Thread.sleep(20);
					continue;
				}
				// 3.解析消息中的订单消息
				MapRecord<String, Object, Object> record = list.get(0);
				Map<Object, Object> values = record.getValue();
				VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
				// 4.如果获取成功, 可以下单
				handleVoucherOrder(voucherOrder);
				// 5.ACK确认 SACK streams.order g1 id
				stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
			} catch (Exception e) {
				log.error("处理订单异常", e);
				handlePendingList();
			}
		}
	}
  1. 这是异步下单的线程执行的逻辑
  2. 不断尝试从消息队列中获取订单信息进行处理
  3. Consumer.from("g1", "c1")创建一个消费者对象, g1消费者组, c1消费者
  4. StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2))

一次读取一条记录, 读取数据时最多等待2秒钟, 超时则返回可读取的数据

  1. StreamOffset.create(queueName, ReadOffset.lastConsumed())

创建一个StreamOffset对象, 从消息队列的上一个未被消费的位置开始读

  1. 解析消息队列中的订单消息
  2. BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true)

将消息从Map结构转换成VoucherOrder对象, 便于后面的使用

  1. handleVoucherOrder(voucherOrder)获取订单信息成功, 调用另一个方法进行下单
  2. stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId())

进行ACK确认, 确认消息队列中的这个订单信息已被消费, 防止重复获取后进行重复下单

  1. handlePendingList()如果出现异常为ACK确认则由pending-list进行处理
/**
 * 处理pending-list中的订单
 */
private void handlePendingList() {
	while (true) {
		try {
			// 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS streams.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()) {
				break;
			}
			// 3.解析消息中的订单消息
			MapRecord<String, Object, Object> record = list.get(0);
			Map<Object, Object> values = record.getValue();
			VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
			// 4.如果获取成功, 可以下单
			handleVoucherOrder(voucherOrder);
			// 5.ACK确认 SACK streams.order g1 id
			stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
		} catch (Exception e) {
			log.error("处理订单异常", e);
			try {
				Thread.sleep(20);
			} catch (InterruptedException interruptedException) {
				interruptedException.printStackTrace();
			}
		}
	}
}
  1. 与上面的处理消息队列的订单信息基本一致, 有些许地方需要改动
  2. 由于是处理异常的消息, 所以只用获取消息队列里的信息来进行处理, 不需要设置阻塞时间
  3. 但拿到的信息标记是0也就是已被消费但为确认ACK的信息
  4. 也就是StreamOffset.create(queueName, ReadOffset.from("0"))
  5. if (list == null || list.isEmpty())如果为空则说明无要处理的信息, 直接结束while循环
  6. 之后的流程与上面基本一致, 如果还出现异常, 不需要再次调用handlePendingList(), 因为有try…catch包裹着, 如果try里面的方法报错的话, 则会输出错误日志后继续进行while循环, 则继续处理异常信息
/**
 * 创建订单
 *
 * @param voucherOrder
 */
@Transactional
@Override
public void createVoucherOrder(VoucherOrder voucherOrder) {
	// 6.一人一单
	Long userId = voucherOrder.getUserId();
	// 6.1.查询订单
	Long count = lambdaQuery()
			.eq(VoucherOrder::getUserId, userId)
			.eq(VoucherOrder::getVoucherId, voucherOrder.getId())
			.count();
	// 6.2.判断是否存在
	if (count > 0) {
		log.error("每人限购一张");
		return;
	}
	boolean success = seckillVoucherService.update()
			.setSql("stock = stock - 1")// set stock = stock - 1
			.eq("voucher_id", voucherOrder.getVoucherId())
			.gt("stock", 0) // where id = ? and stock > 0
			.update();
	if (!success) {
		// 扣减失败
		log.error("库存不足");
		return;
	}
	// 7.创建订单
	save(voucherOrder);
}
  1. 最后则是创建订单, 对数据库进行操作, 根据用户ID和订单ID查询数据库看是否存在订单
  2. 做一次的判断, 防止重复下单, 最后就是修改数据库的订单库存和添加订单信息

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1863397.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

2024年河北省特岗教师报名流程详细图解

2024年河北省特岗教师报名流程详细图解

最近有很多学员们问特岗教师具体的报名流程 给大家安排! 特岗报名步骤 第步: 电脑搜索“河北特岗招聘”登录进行注册 第步:注册后重新登录 第步: 根据个人情况选择填写自己的学历 第步:填写个人信息 (需要上传的电子版的照片、普通话证、学历证书、教资证等) 第步:选择岗位报名…
阅读更多...
【源码+文档+调试讲解】企业人才引进服务平台

【源码+文档+调试讲解】企业人才引进服务平台

摘 要 随着信息时代的来临&#xff0c;过去的传统管理方式缺点逐渐暴露&#xff0c;对过去的传统管理方式的缺点进行分析&#xff0c;采取计算机方式构建企业人才引进服务平台。本文通过课题背景、课题目的及意义相关技术&#xff0c;提出了一种企业信息、招聘信息、应聘信息等…
阅读更多...
敏捷开发笔记(第8章节)--单一职责原则（SRP）

敏捷开发笔记(第8章节)--单一职责原则（SRP）

1&#xff1a;PDF上传链接 【免费】敏捷软件开发(原则模式与实践)资源-CSDN文库 这条原则曾经在Tom DeMaro和Meilir Page-Jones的著作中描述过&#xff0c;并称之为内聚性。他们把内聚性定义为&#xff1a;一个模块的组成元素之间的功能相关性。 8.1 单一职责原则&#xff08…
阅读更多...
【面试干货】Java中==和equals()的区别

【面试干货】Java中==和equals()的区别

【面试干货】Java中和equals&#xff08;&#xff09;的区别 1、操作符2、equals()方法3、总结 &#x1f496;The Begin&#x1f496;点点关注&#xff0c;收藏不迷路&#x1f496; 在Java中&#xff0c;和equals()是两个常用的比较操作符和方法&#xff0c;但它们之间的用法和…
阅读更多...
制图工具（13）地理数据库初始化工具

制图工具（13）地理数据库初始化工具

一、需求背景 地理数据库库体初始化 作为GIS数据管理者&#xff0c;当你拿到数据库表结构&#xff0c;需要你创建一个数据库&#xff1f; 你需要将几个地理数据库的属性结构进行组合、修改&#xff0c;提供一个库体结构&#xff1f; 将不同作业单位&#xff0c;不同作业人员…
阅读更多...
图神经网络实战（15）——SEAL链接预测算法

图神经网络实战（15）——SEAL链接预测算法

图神经网络实战&#xff08;15&#xff09;——SEAL链接预测算法 0. 前言1. SEAL 框架1.1 基本原理1.2 算法流程 2. 实现 SEAL 框架2.1 数据预处理2.2 模型构建与训练 小结系列链接 0. 前言 我们已经学习了基于节点嵌入的链接预测算法&#xff0c;这种方法通过学习相关的节点嵌…
阅读更多...
【第三方JSON库】org.json.simple用法初探—Java编程【Eclipse平台】【不使用项目管理工具】【不添加依赖解析】

【第三方JSON库】org.json.simple用法初探—Java编程【Eclipse平台】【不使用项目管理工具】【不添加依赖解析】

本文将重点介绍&#xff0c;在不使用项目管理工具&#xff0c;不添加依赖解析情况下&#xff0c;【第三方库】JSON.simple库在Java编程的应用。 JSON.simple是一种由纯java开发的开源JSON库&#xff0c;包含在JSON.simple.jar中。它提供了一种简单的方式来处理JSON数据和以JSO…
阅读更多...
SQL Server 2022从入门到精通

SQL Server 2022从入门到精通

大家好&#xff0c;我是爱编程的喵喵。双985硕士毕业&#xff0c;现担任全栈工程师一职&#xff0c;热衷于将数据思维应用到工作与生活中。从事机器学习以及相关的前后端开发工作。曾在阿里云、科大讯飞、CCF等比赛获得多次Top名次。现为CSDN博客专家、人工智能领域优质创作者。…
阅读更多...
架构是怎样练成的-楼宇监控系统案例

架构是怎样练成的-楼宇监控系统案例

目录 概要 项目背景 原系统设计方案 改进后的设计方案 小结 概要 绝大多数人掌握的架构都是直接学习&#xff0c;慢慢地才能体会到一个架构的好处。架构是一种抽象&#xff0c;是为了复用目的而对代码做的抽象。通过一个项目的改造&#xff0c;理解架构是如何产生的&…
阅读更多...
[C++][设计模式][抽象工厂]详细讲解

[C++][设计模式][抽象工厂]详细讲解

目录 1.动机2.模式定义3.要点总结4.代码感受1.代码一2.代码二 -- 工厂方法3.代码三 -- 抽象工厂 1.动机 在软件系统中&#xff0c;经常面临着“一系列相互依赖的对象”的创建工作&#xff1b;同时&#xff0c;由于需求的变化&#xff0c;往往存在更多系列对象的创建工作如何应…
阅读更多...
【ARM】MDK工程切换高版本的编译器后出现error A1137E报错

【ARM】MDK工程切换高版本的编译器后出现error A1137E报错

【更多软件使用问题请点击亿道电子官方网站】 1、 文档目标 解决工程从Compiler 5切换到Compiler 6进行编译时出现一些非语法问题上的报错。 2、 问题场景 对于一些使用Compiler 5进行编译的工程&#xff0c;要切换到Compiler 6进行编译的时候&#xff0c;原本无任何报错警告…
阅读更多...
Redis-哨兵模式-主机宕机-推选新主机的过程

Redis-哨兵模式-主机宕机-推选新主机的过程

文章目录 1、为哨兵模式准备配置文件2、启动哨兵3、主机6379宕机3.4、查看sentinel控制台日志3.5、查看6380主从信息 4、复活63794.1、再次查看sentinel控制台日志 1、为哨兵模式准备配置文件 [rootlocalhost redis]# ll 总用量 244 drwxr-xr-x. 2 root root 150 12月 6 2…
阅读更多...
免费APP分发平台：小猪APP分发如何解决开发者的痛点

免费APP分发平台：小猪APP分发如何解决开发者的痛点

你是否曾为自己开发的APP找不到合适的分发平台而烦恼&#xff1f;你是否因为高昂的分发费用而望而却步&#xff1f;放心吧&#xff0c;你并不是一个人。很多开发者都面临同样的问题。但别担心&#xff0c;小猪APP分发来了&#xff0c;它可以帮你解决这些问题。 小猪app封装www…
阅读更多...
微软结束将数据中心置于海底的实验

微软结束将数据中心置于海底的实验

2016 年&#xff0c;微软 宣布了一项名为"纳蒂克项目"&#xff08;Project Natick&#xff09;的实验。基本而言&#xff0c;该项目旨在了解数据中心能否在海洋水下安装和运行。经过多次较小规模的测试运行后&#xff0c;该公司于 2018 年春季在苏格兰海岸外 117 英尺…
阅读更多...
《Redis设计与实现》阅读总结-2

《Redis设计与实现》阅读总结-2

第 7 章 压缩列表 1. 概念&#xff1a; 压缩列表是列表键和哈希键的底层实现之一。当一个列表键只包含少量列表项&#xff0c;并且每个列表项是小整数值或长度比较短的字符串&#xff0c;那么Redis就会使用压缩类别来做列表键的底层实现。哈希键里面包含的所有键和值都是最小…
阅读更多...
基于ESP8266串口WIFI模块ESP-01S在AP模式（即发射无线信号（ WiFi））下实现STC单片机与手机端网路串口助手相互通信功能

基于ESP8266串口WIFI模块ESP-01S在AP模式（即发射无线信号（ WiFi））下实现STC单片机与手机端网路串口助手相互通信功能

基于ESP8266串口WIFI模块ESP-01S在AP模式（即发射无线信号（ WiFi））下实现STC单片机与手机端网路串口助手相互通信功能 ESP8266_01S引脚功能图ESP8266_01S原理图ESP8266_01S尺寸图检验工作1、USB-TTL串口工具（推荐使用搭载CP2102芯片的安信可USB-T1串口）与ESP8266_01S WiFi…
阅读更多...
Websocket在Java中的实践——最小可行案例

Websocket在Java中的实践——最小可行案例

WebSocket是一种先进的网络通信协议&#xff0c;它允许在单个TCP连接上进行全双工通信&#xff0c;即数据可以在同一时间双向流动。WebSocket由IETF标准化为RFC 6455&#xff0c;并且已被W3C定义为JavaScript API的标准&#xff0c;成为现代浏览器的重要特性之一。 WebSocket的…
阅读更多...
代码随想录——跳跃游戏（Leecode55）

代码随想录——跳跃游戏（Leecode55）

题目链接 贪心 class Solution {public boolean canJump(int[] nums) {int cover 0;if(nums.length 1){return true;}// 只有一个元素可以达到for(int i 0; i < cover; i){// 在cover内选择跳跃步数cover Math.max(i nums[i],cover);if(cover > nums.length - 1)…
阅读更多...
C++进修——C++核心编程

C++进修——C++核心编程

内存分区模型 C程序在执行时&#xff0c;将内存大方向划分为4个区域 代码区&#xff1a;存放函数体的二进制编码&#xff0c;由操作系统进行管理全局区&#xff1a;存放全局变量和静态变量以及常量栈区&#xff1a;由编译器自动分配释放&#xff0c;存放函数的参数值&#xff…
阅读更多...
如何在FastAPI服务器中添加黑名单和白名单实现IP访问控制

如何在FastAPI服务器中添加黑名单和白名单实现IP访问控制

文章目录 📖 介绍 📖🏡 演示环境 🏡📒 文章内容 📒📝 添加黑名单功能步骤1：安装依赖步骤2：创建FastAPI应用步骤3：添加黑名单📝 添加白名单功能步骤1：创建白名单列表步骤2：添加白名单检查⚓️ 相关链接 ⚓️📖 介绍 📖 在现代网络应用开发中，为了增强…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023