文章目录
- 前言
- 4.5 分布式锁-Redisson
- 4.5.4 Redission锁重试
- 4.5.5 WatchDog机制
- 4.5.5 MutiLock原理
- 4.6 秒杀优化
- 4.6.1 优化方案
- 4.6.2 完成秒杀优化
- 4.7 Redis消息队列
- 4.7.1 基于List实现消息队列
- 4.7.2 基于PubSub的消息队列
- 4.7.3 基于Stream的消息队列
- 4.7.4 基于Stream的消息队列-消费者组
- 4.7.5 基于Stream的消息队列实现异步秒杀下单
前言
Redis实战系列文章：
Redis从入门到精通(四)Redis实战(一)短信登录
Redis从入门到精通(五)Redis实战(二)商户查询缓存
Redis从入门到精通(六)Redis实战(三)优惠券秒杀
Redis从入门到精通(七)Redis实战(四)库存超卖、一人一单与Redis分布式锁
Redis从入门到精通(八)Redis实战(五)分布式锁误删与原子性问题、Redisson
4.5 分布式锁-Redisson
上一节对Redisson进行了快速入门，并分析了可重入锁的基本原理，下面继续研究一些Redisson的几个功能。
4.5.4 Redission锁重试
// org.redisson.RedissonLock#lock()
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// 返回null表示获取锁成功，否则返回锁的剩余生存时间
if (ttl == null) {
return;
}
// ......
// 重试获取锁
while (true) {
ttl = tryAcquire(-1, leaseTime, unit, threadId);
if (ttl == null) {
break;
}
// ......
}
由以上源码可知，在RedissonLock类的lock()方法中，会调用tryAcquire()方法尝试获取锁。tryAcquire()方法的原理在上一节已经分析过，返回null表示获取锁成功，否则返回锁的剩余生存时间。
如果第一次获取锁失败，程序会进入一个while循环，重试获取锁。
4.5.5 WatchDog机制
// org.redisson.RedissonLock
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
if (leaseTime != -1) {
return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
// 调用Lua脚本
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// 执行看门狗机制
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
private void scheduleExpirationRenewal(long threadId) {
// ......
} else {
entry.addThreadId(threadId);
// 执行看门狗
renewExpiration();
}
}
private void renewExpiration() {
// ......
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
// ......
// 调用Lua脚本刷新锁的有效时间
RFuture<Boolean> future = renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
// logger
return;
}
if (res) {
// 递归执行看门狗
renewExpiration();
}
});
}
// 10s执行一次
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
由以上源码可知，在RedissonLock类的tryAcquireAsync()方法中，除了调用Lua脚本获取锁，还会运行看门狗机制。该机制会调用Lua脚本刷新锁的有效时间，同时每10s递归执行一次看门狗。
4.5.5 MutiLock原理
为了提高Redis的可用性，一般会搭建集群或者主从。
以主从为例，此时要去获取锁，命令写在主机上，主机会将数据同步给从机。假设在主机还没有来得及把数据写入到从机去的时候，主机宕机了，哨兵会发现主机宕机，并且选举一个Slave变成Master，但此时新的Master中实际上并没有锁信息，相当于此时锁信息已经丢掉了。
为了解决这个问题，Redission提出来了MutiLock锁，使用这种锁后每个节点的地位都是一样的，加锁的逻辑需要把数据写入到每一个主丛节点上，只有所有的节点都写入成功，此时才是真的加锁成功。
假设现在某个节点挂了，那么去获得锁的时候，有一个节点拿不到，不能算是加锁成功，就保证了加锁的可靠性。
4.6 秒杀优化
4.6.1 优化方案
- 现存问题
如上图所示，秒杀下单包括六个步骤：查询优惠券、判断秒杀库存、查询订单、校验一人一单、减库存、创建订单。
在这六步操作中，有很多操作是要去操作数据库的，而且还是一个线程串行执行， 这样就会导致程序执行的很慢。 那么如何加速呢？
- 优化方案
把简单的校验（例如是否有库存、是否一人一单）做完后，就直接给用户返回成功或失败，而不必等待订单创建完成。如果确定可以下单，则将订单的相关信息写入队列，然后再创建一个线程，让新线程读取队列信息异步进行下单。 如下图所示：
- 整体思路
当用户下单时，首先通过Redis判断库存是否充足，如果不充足则直接返回失败；充足的话，再通过Redis判断用户是否已经下过单，如果已经下过单，则直接返回失败；如果没有下过单，则说明可以下单，进行库存扣减，并将用户ID存入当前优惠券的集合中。由于以上过程需要保证原子性，因此可以通过Lua脚本来完成。可以成功下单，Lua脚本返回0。
接着判断Lua脚本的执行结果。如果Lua脚本返回0，说明可以下单，则将优惠券ID、用户ID和订单ID存入阻塞队列，并返回订单ID给用户；如果Lua脚本没有返回0，则直接返回错误信息给用户。
最后进行异步下单，即通过额外线程读取阻塞队列的信息并真正进行下单。完整的流程如下图所示。
4.6.2 完成秒杀优化
- 需求1：新增秒杀优惠券的同时，将优惠券信息保存到Redis中
// com.star.redis.dzdp.service.impl.VoucherServiceImpl
@Override
public BaseResult addSeckillVoucher(Voucher voucher) {
log.info("add a seckill voucher, {}", voucher.toString());
// 1.保存优惠券信息
save(voucher);
log.info("add voucher success. id = {}", voucher.getId());
// 2.保存秒杀信息
SeckillVoucher seckillVoucher = new SeckillVoucher();
seckillVoucher.setVoucherId(voucher.getId());
seckillVoucher.setStock(voucher.getStock());
seckillVoucher.setBeginTime(voucher.getBeginTime());
seckillVoucher.setEndTime(voucher.getEndTime());
seckillVoucherService.save(seckillVoucher);
// 3.将秒杀优惠券的库存保存到Redis
String key = "seckill:stock:" + voucher.getId();
stringRedisTemplate.opsForValue().set(key, voucher.getStock().toString());
log.info("set to Redis : Key = {}, Value = {}", key, voucher.getStock().toString());
return BaseResult.setOk("新增秒杀券成功！");
}
调用/voucher/seckill/order接口新增一个描述优惠券：
在Redis中可以看到该秒杀优惠券的库存信息：
- 需求2：基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
在resources目录下新建一个order.lua文件，其内容如下：
-- 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.库存不足，返回1
return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
-- 3.3.存在，说明是重复下单，返回2
return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
return 0
- 需求3：如果抢购成功，将优惠券ID、用户ID和订单ID封装后存入阻塞队列
修改VoucherOrderServiceImpl类的下单方法seckillVoucher()：
// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl
/** 保存订单信息的队列 */
private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
@Override
public BaseResult<Long> seckillVoucher(Long voucherId, Long userId) {
log.info("开始秒杀下单...voucherId = {}, userId = {}", voucherId, userId);
Long orderId = RedisIdWorker.nextId(stringRedisTemplate, "voucher_order");
log.info("get orderId = {}", orderId);
// 1.执行Lua脚本
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
script.setLocation(new ClassPathResource("order.lua"));
script.setResultType(Long.class);
Long result = stringRedisTemplate.execute(script, Collections.emptyList(),
voucherId.toString(), userId.toString(), orderId.toString());
log.info("execute order.lua result = {}", result);
// 2.判断执行结果
if(result == null || result != 0) {
// 结果为空或者不为0
String message = (result == null || result == 1) ? "库存不足" : "不能重复下单";
log.error(message);
return BaseResult.setFail(message);
}
// 3.结果为0，将优惠券ID、用户ID和订单ID封装后存入阻塞队列
VoucherOrder voucherOrder = new VoucherOrder();
voucherOrder.setVoucherId(voucherId);
voucherOrder.setUserId(userId);
voucherOrder.setId(orderId);
orderTasks.add(voucherOrder);
log.info("add voucherId = {}, userId = {}, orderId = {} to queue.. done.",
voucherId, userId, orderId);
// 4.返回订单ID
log.info("秒杀下单返回...orderId = {}", orderId);
return BaseResult.setOkWithData(orderId);
}
- 需求4：开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能
// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl
/** 异步执行下单动作的线程池 */
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循环持续读取队列中的信息
while (true) {
try {
log.info("=====begin=====>");
// 1.获取队列中的订单信息
VoucherOrder voucherOrder = orderTasks.take();
log.info("get from queue : {}", voucherOrder.toString());
// 2.创建订单
handleVoucherOrder(voucherOrder);
log.info("=====end=====>");
} catch (Exception e) {
log.error("处理异常订单", e);
}
}
}
/**
* 处理订单
*/
private void handleVoucherOrder(VoucherOrder voucherOrder) {
// 1.创建锁对象
RLock rLock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId());
// 2.尝试获取锁
boolean isLock = rLock.tryLock();
log.info("isLock = {}", isLock);
// 3.判断是否获取锁成功
if(!isLock) {
// 获取锁失败
log.error("不允许重复下单！");
return;
}
try {
// 4.持锁真正创建订单
checkAndCreateVoucherOrder(voucherOrder.getVoucherId(), voucherOrder.getUserId());
} finally {
// 5.释放锁
rLock.unlock();
log.info("unlock done.");
}
}
/**
* 持锁真正创建订单
*/
private void createVoucherOrder(VoucherOrder voucherOrder) {
log.info("begin createVoucherOrder... voucherId = {}, userId = {}, orderId = {}",
voucherOrder.getVoucherId(), voucherOrder.getUserId(), voucherOrder.getId());
// 1.增加一人一单规则
int count = query().eq("voucher_id", voucherOrder.getVoucherId())
.eq("user_id", voucherOrder.getUserId()).count();
log.info("old order count = {}", count);
if(count > 0) {
// 该用户已下过单
log.error("每个帐号只能抢购一张优惠券！");
return;
}
// 2.扣减库存
boolean update = seckillVoucherService.update().setSql("stock = stock - 1")
.eq("voucher_id", voucherOrder.getVoucherId())
.gt("stock", 0)
.update();
log.info("update result = {}", update);
if(!update) {
// 扣减库存失败，返回抢券失败
log.error("库存不足，抢券失败！");
return;
}
// 3.创建订单
voucherOrder.setPayTime(new Date());
voucherOrderService.save(voucherOrder);
}
}
下面借助工具对秒杀下单接口进行性能测试，结果如下：
由于使用的是同一用户，因此971个请求中，只有一个请求是成功的，其余的请求都失败。查看此时Redis中的订单数据，只有1条：
4.7 Redis消息队列
如上图所示，最简单的消息队列包含3个角色：
- 消息队列：存储和管理消息，也被称为消息代理（Message Broker）；
- 生产者：发送消息到消息队列；
- 消费者：从消息队列获取消息并处理消息。
使用队列的好处在于解耦。 在秒杀下单中，用户下单之后，利用Redis去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快了响应速度。
4.7.1 基于List实现消息队列
Redis的List数据结构是一个双向链表，很容易模拟出队列效果。我们可以利用：LPUSH结合RPOP、或者RPUSH结合LPOP来实现。
不过要注意的是，当队列中没有消息时，RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。如图：
基于List的消息队列有哪些优缺点？
优点：
- 利用Redis存储，不受限于JVM内存上限；
- 基于Redis的持久化机制，数据安全性有保证；
- 可以满足消息有序性。
缺点：
- 无法避免消息丢失；
- 只支持单消费者。
4.7.2 基于PubSub的消息队列
PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。如图：
主要命令有：
# 订阅一个或多个频道
SUBSCRIBE channel [channel]
# 订阅与pattern格式匹配的所有频道
PSUBSCRIBE pattern[pattern]
# 向一个频道发送消息
PUBLISH channel msg
基于PubSub的消息队列有哪些优缺点？
优点：
- 采用发布订阅模型，支持多生产、多消费。
缺点：
- 不支持数据持久化；
- 无法避免消息丢失；
- 消息堆积有上限，超出时数据丢失。
4.7.3 基于Stream的消息队列
Stream是Redis 5.0引入的一种新数据类型，可以实现一个功能非常完善的消息队列。
发送消息的命令是：
例如：
127.0.0.1:6379> XADD users * name Rose age 22
"1712458704764-0"
127.0.0.1:6379> XADD users * name Jack age 30
"1712458778623-0"
读取消息的方式之一：XREAD
例如，使用XREAD读取第一个消息：
127.0.0.1:6379> XREAD COUNT 1 STREAMS users 0
1) 1) "users"
2) 1) 1) "1712458704764-0"
2) 1) "name"
2) "Rose"
3) "age"
4) "22"
XREAD阻塞方式，读取最新消息：
# 阻塞1秒
127.0.0.1:6379> XREAD COUNT 1 BLOCK 1000 STREAMS users $
(nil)
(1.02s)
基于STREAM的消息队列的特点：
- 消息可回溯；
- 一个消息可以被多个消费者读取；
- 可以阻塞读取；
- 有消息漏读的风险。
4.7.4 基于Stream的消息队列-消费者组
消费者组（Consumer Group），就是将多个消费者划分到一个组中，监听同一个队列。它具备下列特点：
创建消费者组：
127.0.0.1:6379> XGROUP CREATE users a_group 0
OK
给自定的消费者组添加消费者：
127.0.0.1:6379> XGROUP CREATECONSUMER users a_group a_consumer1
(integer) 1
从消费者组读取消息：
127.0.0.1:6379> XREADGROUP GROUP a_group a_consumer1 COUNT 1 STREAMS users 0
1) 1) "users"
2) (empty array)
基于STREAM消费者组的消息队列的特点：
- 消息可回溯；
- 可以多消费者争抢消息，加快消费速度；
- 可以阻塞读取；
- 没有消息漏读的风险；
- 有消息确认机制，保证消息至少被消费一次。
下面，对比一下这4种消息队列的特点：
经过比较，本案例选择使用基于Stream的消息队列来实现异步秒杀下单。
4.7.5 基于Stream的消息队列实现异步秒杀下单
- 修改秒杀下单Lua脚本order.lua，在认定有抢购资格后，直接向stream.orders队列中添加消息，内容包含voucherId、userId、orderId
-- ...
-- 新增逻辑
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0
- 修改消息读取策略，改为读取Redis的Stream结构队列
// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl#seckillVoucher()
// ......
// 3.结果为0，将优惠券ID、用户ID和订单ID封装后存入阻塞队列
// 新逻辑：这里不再保存队列，在lua脚本中保存
// VoucherOrder voucherOrder = new VoucherOrder();
// voucherOrder.setVoucherId(voucherId);
// voucherOrder.setUserId(userId);
// voucherOrder.setId(orderId);
// orderTasks.add(voucherOrder);
// log.info("add voucherId = {}, userId = {}, orderId = {} to queue.. done.",
// voucherId, userId, orderId);
// 4.返回订单ID
log.info("秒杀下单返回...orderId = {}", orderId);
return BaseResult.setOkWithData(orderId);
// com.star.redis.dzdp.service.impl.VoucherOrderServiceImpl
private class VoucherOrderHandler implements Runnable {
@Override
public void run() {
// 持续读取队列中的信息
while (true) {
try {
log.info("=====begin=====>");
// 1.获取队列中的订单信息
// VoucherOrder voucherOrder = orderTasks.take();
// log.info("get from queue : {}", voucherOrder.toString());
// 1.新逻辑：读取Redis的Stream消息队列
// XREADGROUP GROUP a_group a_consumer1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("a_group", "a_consumer1"),
StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
StreamOffset.create("stream.orders", ReadOffset.lastConsumed()));
// 2.判断订单信息是否为空
if(list == null || list.isEmpty()) {
// 如果为空，说明没有消息，继续下一次循环
continue;
}
// 3.解析数据
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> value = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
log.info("get from Redis Stream queue : id = {}, {}", record.getId(), voucherOrder.toString());
// 4.创建订单
handleVoucherOrder(voucherOrder);
// 5.确认消息
stringRedisTemplate.opsForStream().acknowledge("stream.orders", "a_group", record.getId());
log.info("ack message done.");
log.info("=====end=====>");
} catch (Exception e) {
log.error("处理异常订单", e);
}
}
}
// ......
}
测试：
[http-nio-8081-exec-2] 开始秒杀下单...voucherId = 15, userId = 1012
[http-nio-8081-exec-2] get orderId = 7354966481756487681
[http-nio-8081-exec-2] execute order.lua result = 0
[http-nio-8081-exec-2] add voucherId = 15, userId = 1012, orderId = 7354966481756487681 to queue.. done.
[http-nio-8081-exec-2] 秒杀下单返回...orderId = 7354966481756487681
// 创建新线程异步处理下单逻辑
// 成功获取到Stream队列的消息
[pool-2-thread-1] get from Redis Stream queue : id = 1712461578801-0, VoucherOrder(id=7354966481756487681, userId=1012, voucherId=15, payType=null, status=null, createTime=null, payTime=null, useTime=null, refundTime=null, updateTime=null)
[pool-2-thread-1] isLock = true
[pool-2-thread-1] begin createVoucherOrder... voucherId = 15, userId = 1012, orderId = 7354966481756487681
[pool-2-thread-1] ==> Preparing: SELECT COUNT( * ) FROM tb_voucher_order WHERE (voucher_id = ? AND user_id = ?)
[pool-2-thread-1] ==> Parameters: 15(Long), 1012(Long)
[pool-2-thread-1] <== Total: 1
[pool-2-thread-1] old order count = 0
[pool-2-thread-1] ==> Preparing: UPDATE tb_seckill_voucher SET stock = stock - 1 WHERE (voucher_id = ? AND stock > ?)
[pool-2-thread-1] ==> Parameters: 15(Long), 0(Integer)
[pool-2-thread-1] <== Updates: 1
[pool-2-thread-1] update result = true
[pool-2-thread-1] ==> Preparing: INSERT INTO tb_voucher_order ( id, user_id, voucher_id, pay_time ) VALUES ( ?, ?, ?, ? )
[pool-2-thread-1] ==> Parameters: 7354966481756487681(Long), 1012(Long), 15(Long), 2024-04-07 11:46:21.208(Timestamp)
[pool-2-thread-1] <== Updates: 1
[pool-2-thread-1] unlock done.
// 消息确认完成
[pool-2-thread-1] ack message done.
可见，基于Stream的消息队列正常工作。
…
本节完，更多内容请查阅分类专栏：Redis从入门到精通
感兴趣的读者还可以查阅我的另外几个专栏：
- SpringBoot源码解读与原理分析(已完结)
- MyBatis3源码深度解析(已完结)
- 再探Java为面试赋能(持续更新中…)
