高并发秒杀“零超卖”解决方案：Redisson + Kafka + MySQL 最终一致性实践

高并发秒杀“零超卖”解决方案Redisson Kafka MySQL 最终一致性实践本文档涵盖从理论到代码的完整实现用于解决电商秒杀场景下的库存超卖问题保证数据最终一致性。目录背景与挑战整体架构模式核心流程时序代码实现4.1 Redis Lua 脚本预加载4.2 核心下单服务Redisson 锁 本地消息表4.3 Kafka 消费者MySQL 乐观锁扣减4.4 补偿定时任务保证消息可靠4.5 凌晨对账任务最终一致性修复数据一致性保障机制关键配置参考注意事项与防坑指南总结1. 背景与挑战在高并发秒杀场景下核心痛点在于并发冲突大量请求同时修改同一库存记录导致数据库行锁竞争剧烈。超卖风险若不严格保证“检查库存”与“扣减库存”的原子性则会出现库存为负。性能与一致性权衡强一致性如分布式事务性能极差需采用最终一致性方案。2. 整体架构模式我们采用“缓存预扣 异步落库 补偿兜底”的经典架构各组件职责如下组件角色核心作用Redisson分布式锁防止同一用户重复提交防重入降低无效并发Redis Lua流量闸门原子扣减缓存库存拦截大部分超卖请求保护数据库Kafka异步削峰将下单请求异步化平滑流量峰值并保证消息可靠性MySQL最终权威使用乐观锁version或stock num作为最终裁决保证物理库存准确本地消息表可靠性保障保证 Kafka 消息不丢失同时支持幂等消费定时补偿 对账兜底机制处理异常情况如消息丢失、缓存不一致实现最终一致性3. 核心流程时序用户请求→ 获取 Redisson 分布式锁Key userId:productId防止重复点击。执行 Redis Lua原子扣减缓存库存stock:productId。若扣减失败 → 直接返回“库存不足”释放锁。若扣减成功 → 进入下一步。本地事务向 MySQL 插入订单记录和本地消息日志状态0 待发送。发送 Kafka异步若失败不阻塞依赖后续补偿任务。返回用户“下单成功请等待支付”。Kafka 消费者拉取消息开启 MySQL 事务幂等性检查查询消息状态若已处理则跳过。执行乐观锁 SQL 更新物理库存UPDATE product SET stock stock - #{num}, version version 1 WHERE id #{id} AND stock #{num}。若更新成功 → 插入订单详情更新消息状态为“已消费2”提交事务。若更新失败 → 记录失败发送补偿消息将 Redis 库存加回并通知用户下单失败。补偿定时任务每分钟扫描状态为“待发送”或“已发送但未确认”的旧消息重新发送 Kafka。凌晨对账对比 Redis 缓存库存与 MySQL 物理库存若不一致则以 MySQL 为准修复缓存。4. 代码实现环境Spring Boot 3.x MyBatis-Plus Redisson Kafkaspring-kafka以下代码仅展示核心逻辑请按实际业务调整。4.1 Redis Lua 脚本预加载package com.example.seckill.script; import org.springframework.stereotype.Component; Component public class StockLuaScript { // 扣减脚本KEYS[1]库存KeyARGV[1]购买数量 // 返回 1 成功0 失败 public static final String DECREASE_STOCK if redis.call(get, KEYS[1]) tonumber(ARGV[1]) then redis.call(decrby, KEYS[1], ARGV[1]) return 1 else return 0 end; }4.2 核心下单服务Redisson 锁 本地消息表package com.example.seckill.service; import com.alibaba.fastjson.JSON; import com.example.seckill.entity.LocalMessageLog; import com.example.seckill.mapper.LocalMessageLogMapper; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import lombok.extern.slf4j.Slf4j; import java.util.Collections; import java.util.UUID; import java.util.concurrent.TimeUnit; Service Slf4j public class SeckillOrderService { Autowired private RedissonClient redissonClient; Autowired private StringRedisTemplate redisTemplate; Autowired private LocalMessageLogMapper logMapper; Autowired private KafkaTemplateString, String kafkaTemplate; // 下单入口 public String createOrder(Long userId, Long productId, Integer num) { String lockKey lock:seckill: userId : productId; String stockKey stock: productId; RLock lock redissonClient.getLock(lockKey); boolean locked false; try { // 1. 尝试获取锁最多等待0秒持有200毫秒防死等 locked lock.tryLock(0, 200, TimeUnit.MILLISECONDS); if (!locked) { return 请勿重复点击稍后再试; } // 2. Redis Lua 原子扣减 Long result redisTemplate.execute( new DefaultRedisScript(StockLuaScript.DECREASE_STOCK, Long.class), Collections.singletonList(stockKey), num.toString() ); if (result null || result 0) { return 库存不足秒杀失败; } // 3. 构造订单和本地消息日志 String orderId UUID.randomUUID().toString(); LocalMessageLog logEntity new LocalMessageLog(); logEntity.setOrderId(orderId); logEntity.setProductId(productId); logEntity.setUserId(userId); logEntity.setNum(num); logEntity.setStatus(0); // 0待发送1已发送2已消费 // 4. 本地事务保存日志同时保存订单此处省略订单insert // 注意实际中需将 insert 放在 Transactional 方法中 saveOrderAndLog(logEntity); // 内部使用 Transactional // 5. 发送 Kafka异步失败不阻塞 kafkaTemplate.send(seckill-order-topic, orderId, JSON.toJSONString(logEntity)); // 可选异步更新消息状态为1但依赖补偿兜底可省略 return 下单成功订单号 orderId 请等待支付; } catch (Exception e) { // 本地事务失败必须回滚 Redis 库存 log.error(本地事务异常执行Redis回滚, e); redisTemplate.opsForValue().increment(stockKey, num); return 系统繁忙请稍后重试; } finally { if (locked lock.isHeldByCurrentThread()) { lock.unlock(); } } } Transactional(rollbackFor Exception.class) public void saveOrderAndLog(LocalMessageLog logEntity) { // 插入订单表略 // orderMapper.insert(order); logMapper.insert(logEntity); } }4.3 Kafka 消费者MySQL 乐观锁扣减package com.example.seckill.consumer; import com.alibaba.fastjson.JSON; import com.example.seckill.entity.LocalMessageLog; import com.example.seckill.mapper.LocalMessageLogMapper; import com.example.seckill.mapper.OrderMapper; import com.example.seckill.mapper.ProductMapper; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Component; Component Slf4j public class SeckillOrderConsumer { Autowired private OrderMapper orderMapper; Autowired private ProductMapper productMapper; Autowired private LocalMessageLogMapper logMapper; Autowired private KafkaTemplateString, String kafkaTemplate; KafkaListener(topics seckill-order-topic, groupId seckill-group) public void consume(ConsumerRecordString, String record, Acknowledgment ack) { String orderId record.key(); LocalMessageLog logEntity JSON.parseObject(record.value(), LocalMessageLog.class); // 1. 幂等性校验 LocalMessageLog existing logMapper.selectByOrderId(orderId); if (existing null || existing.getStatus() 2) { ack.acknowledge(); return; // 已处理跳过 } // 2. MySQL 乐观锁扣减 int updateRows productMapper.decreaseStockWithOptimisticLock( logEntity.getProductId(), logEntity.getNum() ); // Mapper SQL: // UPDATE product SET stock stock - #{num}, version version 1 // WHERE id #{id} AND stock #{num} if (updateRows 0) { // 扣减成功生成订单 Order order new Order(); order.setOrderId(orderId); order.setUserId(logEntity.getUserId()); order.setStatus(1); // 待支付 orderMapper.insert(order); // 更新消息状态为已消费 logMapper.updateStatus(orderId, 2); ack.acknowledge(); log.info(订单落库成功: {}, orderId); } else { // 扣减失败触发补偿 log.error(物理库存不足订单失败触发补偿: {}, orderId); logMapper.updateStatus(orderId, -1); // 失败状态 // 发送补偿消息将 Redis 库存加回 String compensationMsg {\productId\: logEntity.getProductId() ,\num\: logEntity.getNum() }; kafkaTemplate.send(compensation-topic, orderId, compensationMsg); ack.acknowledge(); // 推送通知用户下单失败略 } } }4.4 补偿定时任务保证消息可靠package com.example.seckill.task; import com.alibaba.fastjson.JSON; import com.example.seckill.entity.LocalMessageLog; import com.example.seckill.mapper.LocalMessageLogMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.util.List; Component Slf4j public class MessageCompensationTask { Autowired private LocalMessageLogMapper logMapper; Autowired private KafkaTemplateString, String kafkaTemplate; // 每5分钟执行一次扫描状态为0待发送或1已发送但未确认且创建时间超过5分钟的消息 Scheduled(cron 0 0/5 * * * ?) public void retryUnsentMessages() { ListLocalMessageLog pendingList logMapper.selectPendingMessages(); // status in (0,1) and create_time now-5min for (LocalMessageLog log : pendingList) { try { kafkaTemplate.send(seckill-order-topic, log.getOrderId(), JSON.toJSONString(log)); // 若发送成功更新状态为1已发送 logMapper.updateStatus(log.getOrderId(), 1); log.info(补偿重发成功: {}, log.getOrderId()); } catch (Exception e) { log.error(补偿重发失败待下次重试: {}, log.getOrderId(), e); } } } }4.5 凌晨对账任务最终一致性修复package com.example.seckill.task; import com.example.seckill.entity.Product; import com.example.seckill.mapper.ProductMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.util.List; Component Slf4j public class StockReconciliationTask { Autowired private ProductMapper productMapper; Autowired private StringRedisTemplate redisTemplate; // 凌晨3点执行 Scheduled(cron 0 0 3 * * ?) public void reconcile() { ListProduct products productMapper.selectAll(); for (Product p : products) { String stockKey stock: p.getId(); String redisVal redisTemplate.opsForValue().get(stockKey); Integer redisStock redisVal null ? 0 : Integer.parseInt(redisVal); if (!redisStock.equals(p.getStock())) { log.error(发现不一致Product: {}, Redis: {}, MySQL: {}, 强制修复, p.getId(), redisStock, p.getStock()); // 以 MySQL 为准覆盖 Redis redisTemplate.opsForValue().set(stockKey, String.valueOf(p.getStock())); // 可发送告警通知人工介入 } } } }5. 数据一致性保障机制为了在异步链路中保证最终一致性我们采用了以下三道防线本地消息表 补偿重试确保 Kafka 消息不丢失即使发送失败也有重试机制。消费幂等通过订单号orderId查询消息状态避免重复消费导致库存多扣。反向补偿当 MySQL 扣减失败时发送补偿消息将 Redis 库存加回并通知用户。定期对账每日凌晨比对 Redis 与 MySQL 库存自动修复差异并记录告警。这套机制保证了在极端情况下如网络分区、服务重启数据最终会趋于一致。6. 关键配置参考application.yml部分spring: kafka: bootstrap-servers: localhost:9092 producer: retries: 3 acks: all consumer: group-id: seckill-group enable-auto-commit: false auto-offset-reset: latest listener: ack-mode: manual redis: host: localhost port: 6379 datasource: url: jdbc:mysql://localhost:3306/seckill?useSSLfalseallowMultiQueriestrue driver-class-name: com.mysql.cj.jdbc.Driver username: root password: 123456本地消息表 DDLCREATE TABLE local_message_log ( id bigint(20) NOT NULL AUTO_INCREMENT, order_id varchar(64) NOT NULL COMMENT 订单号, user_id bigint(20) NOT NULL, product_id bigint(20) NOT NULL, num int(11) NOT NULL, status tinyint(4) DEFAULT 0 COMMENT 0-待发送 1-已发送 2-已消费 -1-失败, create_time datetime DEFAULT CURRENT_TIMESTAMP, update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_order_id (order_id), KEY idx_status_create (status, create_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;7. 注意事项与防坑指南事务边界不要在Transactional中调用 Kafka 发送避免网络抖动导致 DB 事务回滚。应先提交事务再异步发送失败由补偿任务处理。Redisson 锁续期业务执行超过 30 秒时Redisson 会自动续期看门狗无需担心锁提前释放。Kafka 手动提交必须使用Acknowledgment.acknowledge()并关闭自动提交确保消费成功后才提交 Offset防止消息丢失。乐观锁 SQL 条件务必加上stock #{num}这是防超卖的数据库最后防线。Redis 回滚若本地事务DB失败务必立即将 Redis 库存加回否则会造成缓存与 DB 不一致。8. 总结本方案通过Redisson 防重、Redis Lua 防超、Kafka 异步削峰、MySQL 乐观锁兜底、本地消息表保可靠、定时对账修数据构建了一套高并发下零超卖的最终一致性体系。各层职责清晰性能与数据安全得到平衡。实际生产部署时请根据自身业务调整超时参数、重试次数和监控告警以便及时发现并处理异常。