摘要：
本文介绍了基于Redis的秒杀系统优化方案，主要包含两部分：1）通过Lua脚本校验用户秒杀资格，结合Java异步处理订单提升性能；2）使用Redis Stream实现消息队列处理订单。方案采用Lua脚本保证库存校验和一人一单的原子性，通过阻塞队列异步保存订单，并引入Redisson分布式锁防止重复下单。
Redis Stream实现消息队列，支持消费组和ACK确认机制，确保订单可靠处理。系统还设计了pending-list异常处理机制，保证订单处理的最终一致性。这种架构显著提升了秒杀系统的高并发性能和数据一致性。
一，秒杀优化
redis校验用户秒杀资格（库存是否充足且保证一人一单），通过阻塞队列异步处理保存订单到数据库的操作，提升秒杀性能。
1，编写校验秒杀资格lua脚本
库存不足返回1，用户重复下单返回2，资格校验通过返回0。
local voucherId = ARGV[1]
local userId = ARGV[2]
local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId
---库存不足
if tonumber(redis.call('get', 'stockKey'))<=0 then
return 1
end
---库存充足，判断用户是否下单
if redis.call('sismember', orderKey, userId) ==1 then
---重复下单
return 2
end
---扣减库存保存用户id到set中
redis.call('incrby', stockKey, -1)
redis.call('sadd', orderKey, userId)
return 02，读取lua脚本到Java程序
（1）lua文件读取配置
private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
SECKILL_SCRIPT = new DefaultRedisScript<>();
//采用spring的读取文件资源的ClassPathResource
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
SECKILL_SCRIPT.setResultType(Long.class);
}（2）调用stringRedisTemlate的execute方法读取SECKILL_SCRIPT脚本，并传入参数ARGV[..]，因为脚本中不需要KEYS[..]变量，所以这里传入空集合。
Long result = stringRedisTemplate.execute(
SECKILL_SCRIPT,
Collections.emptyList(),
voucherId.toString(), userId.toString()
);3，调用阻塞队列BlockingQueue<VoucherOrder>的数组实现并指定大小，将创建好的订单加入到阻塞队列，方法即可结束，大大提升性能。
private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
VoucherOrder voucherOrder = new VoucherOrder();
//订单id，用户id，优惠卷id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
voucherOrder.setUserId(userId);
voucherOrder.setVoucherId(voucherId);
orderTasks.add(voucherOrder);4，开启独立线程处理将订单写入数据库，加上@PostConstruct注解保证在类初始化之后立即执行
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
@PostConstruct
private void init() {
SECKILL_ORDER_EXECUTOR.submit(() -> {
while (true) {
try {
//获取队列的头部，如果需要则等待直到元素可用为止
VoucherOrder order = orderTasks.take();
handleVoucherOrder(order);
} catch (Exception e) {
log.error("处理订单异常", e);
}
}
});
}5，order传入上锁的方法handleVoucherOrder()，防止同一个用户的多个请求并发产生的问题，保证每个请求（线程）单独执行
private void handleVoucherOrder(VoucherOrder order) {
RLock lock = redissonClient.getLock(LOCK_ORDER_KEY + order.getUserId());
try {
boolean isLock = lock.tryLock();
if (!isLock) {
log.error("操作频繁，请稍后重试！");
return;
}
proxy.createVoucherOrder(order);
} finally {
lock.unlock();
}
}6，编写订单写入数据库的方法createVoucherOrder()并开启事务，这样保证锁的范围比事务范围大，避免出现事务未提交锁提前释放的问题。
@Transactional
public void createVoucherOrder(VoucherOrder order) {
//一人一单判断
Long userId = order.getUserId();
Long voucherId = order.getVoucherId();
long count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
if (count > 0) {
log.error("不能重复下单！");
return;
}
//扣减库存
//这个sql语句是原子性的操作,而LambdaUpdateWrapper表达式不是
boolean success = seckillVoucherService.update().setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock", 0).update();
if (!success) {
log.error("库存不足！");
return;
}
save(order);
}
二，redis实现消息队列
（1）基于list模拟消息队列
（2）基于pubsub的消息队列
（3）基于stream类型的消息队列
基于·redis的stream结构作为消息队列，实现异步秒杀
1，创建STREAM数据stream.order作为阻塞队列和消费组g1
xgroup create stream.order g1 0 mkstream2，lua脚本增加发送消息到队列中的操作和订单id
local orderId = ARGV[3]
redis.call('xadd', 'stream.order', '*', 'userId', userId,
'voucherId', voucherId, 'id', orderId)3，java客户端获取消息队列中的消息
（1）获取消息队列的订单信息，redis命令：XREADGROUP GROUP g1 c1 count 1 block 2000 streams stream.order >（消费者c1，读取数量1，阻塞时间2000毫秒，>表示从当前消费者组中未消费的最新的消息开始读取）
（2）如果返回的结果为空或者list是空集合，说明没有获取到，continue后面操作重新获取；获取成功，取出消息MapRecord，取出订单信息键值对record.getValue()，最后填入VoucherOrder即可
（3）ack确认消息xack stream.order g1 id
while (true) {
try {
//获取消息队列的订单信息XREADGROUP GROUP g1 c1 count 1 block 2000 streams stream.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())
);
//获取失败，进行下一次获取循环
if (list == null || list.isEmpty()) {
continue;
}
//获取成功，处理消息队列中的订单信息
MapRecord<String, Object, Object> record = list.getFirst();
Map<Object, Object> value = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
handleVoucherOrder(voucherOrder);
//ack确认消息xack stream.order g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("处理订单异常", e);
//如果中间出现异常那么就在pending-list中保存订单信息
handlePendingList();
}
}4，如果中间出现异常那么就在pending-list中保存订单信息
（1）获取pending-list中的订单信息，redis命令：XREADGROUP GROUP g1 c1 count 1 streams stream.order 0（0表示从开始位置读取消息）
（2）获取失败，说明pending-list里面没有异常消息，那么直接结束循环；如果再次出现异常，记录日志，休眠线程一段时间 ，重新开始循环。
private void handlePendingList() {
while (true) {
try {
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"),
StreamReadOptions.empty().count(1),
StreamOffset.create(queueName, ReadOffset.from("0"))
);
if (list == null || list.isEmpty()) {
break;
}
MapRecord<String, Object, Object> record = list.getFirst();
//获取订单键值对
Map<Object, Object> value = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
handleVoucherOrder(voucherOrder);
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("处理pending-list异常", e);
try {
Thread.sleep(20);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}
}
}
![[ Qt ] | QPushButton常见用法](https://i-blog.csdnimg.cn/direct/334f8a5a97c54445aae4804912a67fcf.png)
