使用 Redis + MQ 完成抢购活动的最佳实现

好记忆不如烂笔头能记下点东西就记下点有时间拿出来看看也会发觉不一样的感受.一概况1.介绍2.关键信息3.设计思路二核心实现1.添加依赖2.Lua脚本文件3.消息实体4. Redis Lua脚本执行器5. Kafka消息生产者6. 抢购服务实现7. Kafka消费者8. 控制器9. 配置文件 (application.yml)10. 启动类一概况1.介绍有兄弟留言说让我帮忙设计一个秒杀活动目的是拉新大致的情况他是这样告诉我的有个活动用户想参加活动就需要注册成会员然后抢购礼品1000个礼品到账进行消费。以此来拉取新用户和提高用户的活跃度。针对这样的一个抢购场景进行抢购活动的代码设计。2.关键信息核心信息是用户userId活动activityId库存10003.设计思路异步削峰使用MQ 可以采用你熟悉的或者系统中已经存在的MQ资源池加锁使用Redis做抢购资源的管控springboot中推荐使用 Redisson 做分布式锁利用redis的流式处理完成系统的核心代码实现推荐使用lua脚本实现。二核心实现具体的代码实现我就用以下方式进行输出不做过多的阐述直接看代码就好.1.添加依赖dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.kafka/groupId artifactIdspring-kafka/artifactId /dependency dependency groupIdorg.redisson/groupId artifactIdredisson-spring-boot-starter/artifactId version3.23.4/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency /dependencies2.Lua脚本文件-- KEYS[1] 库存键名 -- KEYS[2] 用户已领取集合键名 -- KEYS[3] Stream队列键名 -- ARGV[1] userId -- ARGV[2] 消息内容(JSON格式) -- ARGV[3] 过期时间(秒) -- 检查库存是否充足 local current_stock tonumber(redis.call(GET, KEYS[1])) if not current_stock or current_stock 0 then return {code 0, message 库存不足} end -- 检查用户是否已经领取过 if redis.call(SISMEMBER, KEYS[2], ARGV[1]) 1 then return {code 0, message 您已经参与过此活动} end -- 扣减库存 local new_stock redis.call(DECR, KEYS[1]) redis.call(EXPIRE, KEYS[1], ARGV[3]) -- 记录用户领取 redis.call(SADD, KEYS[2], ARGV[1]) redis.call(EXPIRE, KEYS[2], ARGV[3]) -- 写入Stream队列 redis.call(XADD, KEYS[3], *, userId, ARGV[1], message, ARGV[2]) return {code 1, message 抢购成功, stock new_stock}3.消息实体// GiftPurchaseRequest.java public class GiftPurchaseRequest { private Long userId; private Long activityId; private String giftName; // 构造函数、getter和setter public GiftPurchaseRequest() {} public GiftPurchaseRequest(Long userId, Long activityId, String giftName) { this.userId userId; this.activityId activityId; this.giftName giftName; } // getter和setter方法 public Long getUserId() { return userId; } public void setUserId(Long userId) { this.userId userId; } public Long getActivityId() { return activityId; } public void setActivityId(Long activityId) { this.activityId activityId; } public String getGiftName() { return giftName; } public void setGiftName(String giftName) { this.giftName giftName; } Override public String toString() { return String.format({\userId\: %d, \activityId\: %d, \giftName\: \%s\}, userId, activityId, giftName); } } // GiftPurchaseResult.java public class GiftPurchaseResult { private int code; private String message; private Integer stock; public GiftPurchaseResult(int code, String message, Integer stock) { this.code code; this.message message; this.stock stock; } // getter和setter方法 public int getCode() { return code; } public void setCode(int code) { this.code code; } public String getMessage() { return message; } public void setMessage(String message) { this.message message; } public Integer getStock() { return stock; } public void setStock(Integer stock) { this.stock stock; } }4. Redis Lua脚本执行器Component public class RedisLuaExecutor { Autowired private RedisTemplateString, Object redisTemplate; SuppressWarnings(unchecked) public GiftPurchaseResult executeGiftPurchase(Long userId, String giftInfo, String activityId, int expireSeconds) { try { ListString keys Arrays.asList( gift:stock: activityId, // 库存键 gift:users: activityId, // 已领取用户集合 gift:stream: activityId // Stream队列 ); ListString args Arrays.asList( String.valueOf(userId), // 用户ID giftInfo, // 礼品信息 String.valueOf(expireSeconds) // 过期时间 ); DefaultRedisScriptList redisScript new DefaultRedisScript(); redisScript.setScriptText(luaScript); redisScript.setLocation(new ClassPathResource(lua/gift_purchase.lua)); redisScript.setResultType(List.class); ListObject result (ListObject) redisTemplate.execute( redisScript, new StringRedisSerializer(), new StringRedisSerializer(), keys, args.toArray(new String[0]) ); if (result ! null !result.isEmpty()) { ListObject response (ListObject) result.get(0); int code ((Long) response.get(0)).intValue(); String message (String) response.get(1); Integer stock response.size() 2 ? ((Long) response.get(2)).intValue() : null; return new GiftPurchaseResult(code, message, stock); } return new GiftPurchaseResult(0, 执行失败, null); } catch (Exception e) { e.printStackTrace(); return new GiftPurchaseResult(0, 系统错误: e.getMessage(), null); } } }5. Kafka消息生产者Component public class GiftPurchaseProducer { Autowired private KafkaTemplateString, String kafkaTemplate; Value(${gift.purchase.topic:gift-purchase-topic}) private String topicName; public void sendGiftPurchaseMessage(String userId, String giftInfo) { GiftPurchaseRequest request new GiftPurchaseRequest(Long.valueOf(userId), 10000L, 礼品名称); String message request.toString(); kafkaTemplate.send(topicName, userId, message); } }6. 抢购服务实现Service Slf4j public class GiftPurchaseService { Autowired private RedisLuaExecutor redisLuaExecutor; Autowired private GiftPurchaseProducer giftPurchaseProducer; Autowired private RedissonClient redissonClient; private static final int INITIAL_STOCK 100; private static final int EXPIRE_SECONDS 86400; // 24小时过期 private static final String ACTIVITY_ID 10000; PostConstruct public void initStock() { String stockKey gift:stock: ACTIVITY_ID; Boolean hasStock redisTemplate.hasKey(stockKey); if (!hasStock || (Integer) redisTemplate.opsForValue().get(stockKey) 0) { redisTemplate.opsForValue().set(stockKey, INITIAL_STOCK); redisTemplate.expire(stockKey, EXPIRE_SECONDS, TimeUnit.SECONDS); } } public GiftPurchaseResult purchaseGift(Long userId) { RLock lock redissonClient.getLock(gift_lock_ ACTIVITY_ID); try { boolean acquired lock.tryLock(10, TimeUnit.SECONDS); if (!acquired) { return new GiftPurchaseResult(0, 系统繁忙请稍后再试, null); } String giftInfo String.format({\userId\: %d, \activityId\: %d, \giftName\: \礼品名称\}, userId, 10000L); GiftPurchaseResult result redisLuaExecutor.executeGiftPurchase( userId, giftInfo, ACTIVITY_ID, EXPIRE_SECONDS); if (result.getCode() 1) { // 异步处理礼品发放 giftPurchaseProducer.sendGiftPurchaseMessage(userId.toString(), giftInfo); log.info(用户 {} 抢购成功剩余库存: {}, userId, result.getStock()); } return result; } catch (Exception e) { log.error(抢购过程中发生异常, e); return new GiftPurchaseResult(0, 系统错误: e.getMessage(), null); } finally { if (lock.isHeldByCurrentThread()) { lock.unlock(); } } } }7. Kafka消费者Component Slf4j public class GiftPurchaseConsumer { Value(${gift.purchase.topic:gift-purchase-topic}) private String topicName; KafkaListener(topics ${gift.purchase.topic:gift-purchase-topic}, groupId gift-purchase-group) public void handleGiftPurchase(String message) { try { log.info(开始处理礼品发放消息: {}, message); // 解析JSON消息 ObjectMapper objectMapper new ObjectMapper(); JsonNode jsonNode objectMapper.readTree(message); Long userId jsonNode.get(userId).asLong(); Long activityId jsonNode.get(activityId).asLong(); // 这里执行实际的礼品发放逻辑 // 例如调用外部API、更新数据库等 processGiftDelivery(userId, activityId); log.info(礼品发放成功用户ID: {}, 活动ID: {}, userId, activityId); } catch (Exception e) { log.error(处理礼品发放消息失败: {}, message, e); } } private void processGiftDelivery(Long userId, Long activityId) { // 实际的礼品发放逻辑 // 例如发送邮件、短信通知或者调用第三方礼品发放API System.out.println(正在为用户 userId 发放活动 activityId 的礼品...); } }8. 控制器RestController RequestMapping(/gift) Slf4j public class GiftPurchaseController { Autowired private GiftPurchaseService giftPurchaseService; PostMapping(/purchase/{userId}) public ResponseEntityGiftPurchaseResult purchaseGift(PathVariable Long userId) { log.info(用户 {} 发起抢购请求, userId); GiftPurchaseResult result giftPurchaseService.purchaseGift(userId); return ResponseEntity.ok(result); } GetMapping(/stock/{activityId}) public ResponseEntityInteger getStock(PathVariable String activityId) { String stockKey gift:stock: activityId; Object stockObj redisTemplate.opsForValue().get(stockKey); Integer stock stockObj ! null ? (Integer) stockObj : 0; return ResponseEntity.ok(stock); } }9. 配置文件 (application.yml)spring: redis: host: localhost port: 6379 timeout: 10s kafka: bootstrap-servers: localhost:9092 consumer: group-id: gift-purchase-group auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer gift: purchase: topic: gift-purchase-topic server: port: 808010. 启动类SpringBootApplication EnableKafka public class GiftPurchaseApplication { public static void main(String[] args) { SpringApplication.run(GiftPurchaseApplication.class, args); } }经过上述实现基本可以进行抢购活动的进行当然具体还要根据其具体业务做响应的修改这里只是一个demo只是说明一种思想抢购的实践方案。如此设计的优点是Redis分布式锁: 使用Redisson确保并发安全Lua脚本原子操作: 库存扣减、用户记录、Stream写入一步完成Kafka异步处理: 抢购成功后异步处理礼品发放Redis Stream: 消息持久化和顺序处理防重复领取: 使用Set数据结构防止同一用户多次领取自动过期: Redis键值设置过期时间避免数据堆积