全局ID生成器

---

特性:

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

全局唯一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. 阿巴阿巴阿巴~ 太简单了 自行理解 哈哈哈~

实现优惠券秒杀下单

---

/*** 秒杀优惠券下单* @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.订单IDlong orderId = redisIdWorker.nextId("order");order.setId(orderId);// 6.2.用户IDLong userId = UserHolder.getUser().getId();order.setUserId(userId);// 6.3.代金券IDorder.setVoucherId(voucherId);save(order);// 7.返回订单IDreturn Result.ok(orderId);
}

1. 查询优惠券->判断秒杀时间->判断库存->下单->更新库存
2. 判断秒杀时间要判断起始时间和结束时间, 看是否在时间区间内
3. 判断库存是否充足, 充足则进行库存更新, 使用lambdaUpdate来更新
4. 创建订单, 更新库存, 使用之前的全局唯一ID生成器作为订单ID
5. 在设置用户ID, 秒杀券ID的属性, 最后保存秒杀券信息

超卖问题

---

悲观锁和乐观锁相关文章

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

乐观锁执行逻辑

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博客

一人一单执行逻辑

代码实现

/*** 秒杀优惠券下单* @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.订单IDlong orderId = redisIdWorker.nextId("order");order.setId(orderId);// 7.2.用户IDorder.setUserId(userId);// 7.3.代金券IDorder.setVoucherId(voucherId);save(order);// 8.返回订单IDreturn Result.ok(orderId);
}

1. 一人一单防止超卖问题, 使用悲观锁来限制优惠券的购买
2. 通过悲观锁确保了同一用户创建订单的并发安全性，而使用事务则保证了创建订单的原子性和一致性
3. 在秒杀时间内且库存充足情况下, 会去查询当前用户ID是否已购买当前秒杀券
4. 但短时间大量请求打到数据库上, 查出来的结果部分可能相同, 这依然存在超卖问题
5. 这时单独将查询是否已购买秒杀券和创建订单加上悲观锁, 但此时又存在性能低下
6. 每个线程都要去竞争同一个锁, 出现响应慢的情况, 只用对每个用户加锁即可, 而不是线程加锁
7. synchronized (userId.toString().intern())

• 这是为了保证在多线程环境下，同一用户的订单创建操作不会发生并发冲突
• 通过 synchronized 关键字，使用用户ID的字符串形式作为锁对象进行同步
• 通过调用 toString() 将 userId 转换成字符串, intern() 方法获取字符串的常量池对象，以确保锁的唯一性

6. IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy()

• 这是用来解决前面的事务还没提交而产生的并发问题使事务失效, 导致超卖问题
• 通过 AopContext.currentProxy() 获取当前代理对象，用于在事务中调用 createVoucherOrder 方法
• 这样可以确保事务生效，因为在同一类内调用带有 @Transactional 注解的方法时，默认不会触发事务

集群模式下锁失效

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

分布式锁

---

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

基于Redis的分布式锁

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

• 获取锁:

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

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

• 释放锁:

• 手动释放

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

Redis实现分布式锁流程

实现分布式锁初级版本

@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. 释放锁直接删除锁就行~

分布式锁误删问题

问题现象

解决方案

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

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, 释放锁的时候要进行线程标识和锁标识的比较, 如果标识相同才可以释放

分布式锁的原子性问题

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

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

if(redis.call('get', KEYS[1]) == ARGV[1]) then-- 释放锁 del keyreturn 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获取锁, 并设置过期时间, 保存线程标识
• 释放锁时先判断线程标识是否与自己一致, 一致则删除锁

2. 特性

• 利用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>

2. 创建Redission客户端RedisClient

@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient(){// 配置Config config = new Config();config.useSingleServer().setAddress("redis://192.168.1.112:6379");// 创建RedissonClient对象return Redisson.create(config);}
}

3. 注入RedisClient客户端, 并使用

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

Redission可重入锁原理

1. 获取锁Lua脚本

2. 释放锁Lua脚本

Redisson分布式锁原理

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避免死锁, 释放锁时判断线程标识
• 缺陷: 不可重入, 无法重试, 锁超时失效

2. 可重入的Redis分布式锁

• 原理: 利用Hash结构, 记录线程标识和重入次数, WatchDog延续锁时间, 信号量控制锁重试等待
• 缺陷: redis宕机引起锁失效问题

3. Redisson的MutiLock

• 原理: 多个独立的Redis节点, 必须在所有节点都获取重入锁才算获取锁成功
• 缺陷: 运维成本高, 实现复杂

Redis优化秒杀

---

优化思路

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());
}

2. 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.库存不足，返回1return 1
end
-- 3.2.判断用户是否已下单 sismember orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then-- 3.2.1.已下单，返回2return 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. 判断结果是否为0int r = result.intValue();if(r != 0){// 2.1.不为0, 代表没有购买资格return Result.fail(r == 1 ? "库存不足" : "不可重复下单");}// 2.2.为0, 有购买资格, 把下单信息保存到阻塞队列VoucherOrder voucherOrder = new VoucherOrder();// 2.3.订单IDlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 2.4.用户IDvoucherOrder.setUserId(userId);// 2.5.代金券IDvoucherOrder.setVoucherId(voucherId);// 2.6.放入阻塞队列orderTasks.add(voucherOrder);// 3.获取代理对象proxy = (IVoucherOrderService)AopContext.currentProxy();// 3. 返回订单IDreturn 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{@Overridepublic 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消息队列实现异步秒杀

---

消息队列的定义

基于List结构模拟消息队列

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

优点:

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

缺点:

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

基于PubSub的消息队列

优点:

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

缺点:

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

基于Stream的消息队列

发送消息

读取消息

实现持续监听队列

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

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

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

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

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

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

从消费者读取消息

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

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

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

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

Redis消息队列总结

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

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

内容包括voucherId, userId, orderId

3. 项目启动时, 开启一个线程任务, 尝试获取stream.order中的消息, 完成下单

/*** 秒杀优惠券下单** @param voucherId* @return*/
private IVoucherOrderService proxy;
@Override
public Result seckillVoucher(Long voucherId) {// 获取用户Long userId = UserHolder.getUser().getId();// 获取订单IDlong orderId = redisIdWorker.nextId("order");// 1. 执行lua脚本Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));// 2. 判断结果是否为0int r = result.intValue();if (r != 0) {// 2.1.不为0, 代表没有购买资格return Result.fail(r == 1 ? "库存不足" : "不可重复下单");}// 3.获取代理对象proxy = (IVoucherOrderService) AopContext.currentProxy();// 3. 返回订单IDreturn 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";@Overridepublic 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 idstringRedisTemplate.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秒钟, 超时则返回可读取的数据

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

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

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

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

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

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

10. handlePendingList()如果出现异常为ACK确认则由pending-list进行处理

/*** 处理pending-list中的订单*/
private void handlePendingList() {while (true) {try {// 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS streams.order 0List<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 idstringRedisTemplate.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. 做一次的判断, 防止重复下单, 最后就是修改数据库的订单库存和添加订单信息