【redis实战篇】第六天

摘要：

        本文介绍了基于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 thenreturn 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 0

2，读取lua脚本到Java程序

（1）lua文件读取配置

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();//采用spring的读取文件资源的ClassPathResourceSECKILL_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();@PostConstructprivate 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()并开启事务，这样保证锁的范围比事务范围大，避免出现事务未提交锁提前释放的问题。

    @Transactionalpublic 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 mkstream

2，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 idstringRedisTemplate.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);}}}}