feat: 设计滑动库存分布式锁处理活动秒杀

lottery-application/src/main/java/com/seamew/lottery/application/mq/consumer/LotteryActivityPartakeRecordListener.java

@@ -0,0 +1,50 @@
+package com.seamew.lottery.application.mq.consumer;
+
+import com.alibaba.fastjson2.JSON;
+import com.seamew.lottery.application.mq.producer.KafkaProducer;
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
+import com.seamew.lottery.domain.activity.service.partake.IActivityPartake;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * @Author: seamew
+ * @Title: LotteryActivityPartakeRecordListener
+ * @CreateTime: 2023年03月03日 16:34:00
+ * @Description:
+ * @Version: 1.0
+ */
+@Component
+@Slf4j
+public class LotteryActivityPartakeRecordListener {
+    @Resource
+    private IActivityPartake activityPartake;
+
+    @KafkaListener(topics = KafkaProducer.TOPIC_ACTIVITY_PARTAKE, groupId = KafkaProducer.GROUP_NAME)
+    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
+        // 1. 判断消息是否存在
+        if (ObjectUtils.isEmpty(record.value())) {
+            return;
+        }
+
+        try {
+            // 2. 转化对象
+            ActivityPartakeRecordVO activityPartakeRecordVO = JSON.parseObject((String) record.value(), ActivityPartakeRecordVO.class);
+            log.info("消费MQ消息，异步扣减活动库存 message：{}", activityPartakeRecordVO);
+
+            // 3. 更新数据库库存【实际场景业务体量较大，可能也会由于MQ消费引起并发，对数据库产生压力，所以如果并发量较大，可以把库存记录缓存中，并使用定时任务进行处理缓存和数据库库存同步，减少对数据库的操作次数】
+            activityPartake.updateActivityStock(activityPartakeRecordVO);
+            ack.acknowledge();
+        } catch (Exception e) {
+            // 更新库存失败
+            log.error("更新库存失败--消费MQ消息，失败 topic：{} message：{}", record.topic(), record.value());
+            throw e;
+        }
+    }
+}

lottery-application/src/main/java/com/seamew/lottery/application/mq/consumer/KafkaConsumer.java → lottery-application/src/main/java/com/seamew/lottery/application/mq/consumer/LotteryInvoiceListener.java

@@ -19,19 +19,19 @@ import javax.annotation.Resource;
 
 /**
  * @Author: seamew
- * @Title: KafkaConsumer
+ * @Title: LotteryInvoiceListener
  * @CreateTime: 2023年02月27日 11:12:00
- * @Description: 消息消费者
+ * @Description: 中奖发货单监听消息
  * @Version: 1.0
  */
 @Component
 @Slf4j
-public class KafkaConsumer {
+public class LotteryInvoiceListener {
 
     @Resource
     private DistributionGoodsFactory distributionGoodsFactory;
 
-    @KafkaListener(topics = KafkaProducer.TOPIC_NAME, groupId = KafkaProducer.GROUP_NAME)
+    @KafkaListener(topics = KafkaProducer.TOPIC_INVOICE, groupId = KafkaProducer.GROUP_NAME)
     public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
         // 1. 判断消息是否存在
         if (ObjectUtils.isEmpty(record.value())) {

lottery-application/src/main/java/com/seamew/lottery/application/mq/producer/KafkaProducer.java

@@ -1,13 +1,13 @@
 package com.seamew.lottery.application.mq.producer;
 
 import com.alibaba.fastjson2.JSON;
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
 import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Component;
 import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import javax.annotation.Resource;
 
@@ -24,7 +24,15 @@ public class KafkaProducer {
     @Resource
     private KafkaTemplate<String, Object> kafkaTemplate;
 
-    public static final String TOPIC_NAME = "lottery_invoice";
+    /**
+     * MQ主题：中奖发货单
+     */
+    public static final String TOPIC_INVOICE = "lottery_invoice";
+
+    /**
+     * MQ主题：活动领取记录
+     */
+    public static final String TOPIC_ACTIVITY_PARTAKE = "lottery_activity_partake";
 
     public static final String GROUP_NAME = "lottery";
 
@@ -35,7 +43,18 @@ public class KafkaProducer {
      */
     public ListenableFuture<SendResult<String, Object>> sendLotteryInvoice(InvoiceVO invoice) {
         String objJson = JSON.toJSONString(invoice);
-        log.info("发送MQ消息 topic：{} bizId：{} message：{}", TOPIC_NAME, invoice.getUId(), objJson);
-        return kafkaTemplate.send(TOPIC_NAME, objJson);
+        log.info("发送MQ消息 topic：{} bizId：{} message：{}", TOPIC_INVOICE, invoice.getUId(), objJson);
+        return kafkaTemplate.send(TOPIC_INVOICE, objJson);
+    }
+
+    /**
+     * 发送领取活动记录MQ
+     *
+     * @param activityPartakeRecord 领取活动记录
+     */
+    public void sendLotteryActivityPartakeRecord(ActivityPartakeRecordVO activityPartakeRecord) {
+        String objJson = JSON.toJSONString(activityPartakeRecord);
+        log.info("发送MQ消息(领取活动记录) topic：{} bizId：{} message：{}", TOPIC_ACTIVITY_PARTAKE, activityPartakeRecord.getUId(), objJson);
+        kafkaTemplate.send(TOPIC_ACTIVITY_PARTAKE, objJson);
     }
 }

lottery-application/src/main/java/com/seamew/lottery/application/process/impl/ActivityProcessImpl.java

@@ -9,6 +9,7 @@ import com.seamew.lottery.common.Constants;
 import com.seamew.lottery.common.Result;
 import com.seamew.lottery.domain.activity.model.req.PartakeReq;
 import com.seamew.lottery.domain.activity.model.res.PartakeResult;
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
 import com.seamew.lottery.domain.activity.model.vo.DrawOrderVO;
 import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
 import com.seamew.lottery.domain.activity.service.partake.IActivityPartake;
@@ -54,13 +55,26 @@ public class ActivityProcessImpl implements IActivityProcess {
     public DrawProcessResult doDrawProcess(DrawProcessReq req) {
         // 1. 领取活动
         PartakeResult partakeResult = activityPartake.doPartake(new PartakeReq(req.getUId(), req.getActivityId()));
-        if (!Constants.ResponseCode.SUCCESS.getCode().equals(partakeResult.getCode())) {
+        if (!Constants.ResponseCode.SUCCESS.getCode().equals(partakeResult.getCode()) && !Constants.ResponseCode.NOT_CONSUMED_TAKE.getCode().equals(partakeResult.getCode())) {
             return new DrawProcessResult(partakeResult.getCode(), partakeResult.getInfo());
         }
+
+        // 2. 首次成功领取活动，发送 MQ 消息
+        if (Constants.ResponseCode.SUCCESS.getCode().equals(partakeResult.getCode())) {
+            ActivityPartakeRecordVO activityPartakeRecord = new ActivityPartakeRecordVO();
+            activityPartakeRecord.setUId(req.getUId());
+            activityPartakeRecord.setActivityId(req.getActivityId());
+            activityPartakeRecord.setStockCount(partakeResult.getStockCount());
+            activityPartakeRecord.setStockSurplusCount(partakeResult.getStockSurplusCount());
+            // 发送 MQ 消息
+            kafkaProducer.sendLotteryActivityPartakeRecord(activityPartakeRecord);
+        }
+
+
         Long strategyId = partakeResult.getStrategyId();
         Long takeId = partakeResult.getTakeId();
 
-        // 2. 执行抽奖
+        // 3. 执行抽奖
         DrawResult drawResult = drawExec.doDrawExec(new DrawReq(req.getUId(), strategyId));
         if (Constants.DrawState.FAIL.getCode().equals(drawResult.getDrawState())) {
             // 未中奖也要将记录锁住
@@ -69,14 +83,14 @@ public class ActivityProcessImpl implements IActivityProcess {
         }
         DrawAwardVO drawAwardVO = drawResult.getDrawAwardInfo();
 
-        // 3. 结果落库
+        // 4. 结果落库
         DrawOrderVO drawOrderVO = buildDrawOrderVO(req, strategyId, takeId, drawAwardVO);
         Result recordResult = activityPartake.recordDrawOrder(drawOrderVO);
         if (!Constants.ResponseCode.SUCCESS.getCode().equals(recordResult.getCode())) {
             return new DrawProcessResult(recordResult.getCode(), recordResult.getInfo());
         }
 
-        // 4. 发送MQ，触发发奖流程
+        // 5. 发送MQ，触发发奖流程
         InvoiceVO invoiceVO = buildInvoiceVO(drawOrderVO);
         kafkaProducer
                 .sendLotteryInvoice(invoiceVO)

lottery-common/src/main/java/com/seamew/lottery/common/Constants.java

@@ -15,7 +15,10 @@ public class Constants {
         INDEX_DUP("0003", "主键冲突"),
         NO_UPDATE("0004", "SQL操作无更新"),
         LOSING_DRAW("D001", "未中奖"),
-        RULE_ERR("D002", "量化人群规则执行失败");
+        RULE_ERR("D002", "量化人群规则执行失败"),
+        NOT_CONSUMED_TAKE("D003", "未消费活动领取记录"),
+        OUT_OF_STOCK("D004", "活动无库存"),
+        ERR_TOKEN("D005", "分布式锁失败");
 
         private final String code;
         private final String info;
@@ -45,6 +48,27 @@ public class Constants {
         public static final Long TREE_NULL_NODE = 0L;
     }
 
+    /**
+     * 缓存 Key
+     */
+    public static final class RedisKey {
+
+        // 抽奖活动库存 Key
+        private static final String LOTTERY_ACTIVITY_STOCK_COUNT = "lottery_activity_stock_count_";
+
+        public static String KEY_LOTTERY_ACTIVITY_STOCK_COUNT(Long activityId) {
+            return LOTTERY_ACTIVITY_STOCK_COUNT + activityId;
+        }
+
+        // 抽奖活动库存锁 Key
+        private static final String LOTTERY_ACTIVITY_STOCK_COUNT_TOKEN = "lottery_activity_stock_count_token_";
+
+        public static String KEY_LOTTERY_ACTIVITY_STOCK_COUNT_TOKEN(Long activityId, Integer stockUsedCount) {
+            return LOTTERY_ACTIVITY_STOCK_COUNT_TOKEN + activityId + "_" + stockUsedCount;
+        }
+
+    }
+
     /**
      * 决策树节点
      */

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/model/res/PartakeResult.java

@@ -14,11 +14,18 @@ public class PartakeResult extends Result {
      * 策略ID
      */
     private Long strategyId;
-
     /**
      * 活动领取ID
      */
     private Long takeId;
+    /**
+     * 库存
+     */
+    private Integer stockCount;
+    /**
+     * activity 库存剩余
+     */
+    private Integer stockSurplusCount;
 
     public PartakeResult(String code, String info) {
         super(code, info);
@@ -39,4 +46,20 @@ public class PartakeResult extends Result {
     public void setTakeId(Long takeId) {
         this.takeId = takeId;
     }
+
+    public Integer getStockCount() {
+        return stockCount;
+    }
+
+    public void setStockCount(Integer stockCount) {
+        this.stockCount = stockCount;
+    }
+
+    public Integer getStockSurplusCount() {
+        return stockSurplusCount;
+    }
+
+    public void setStockSurplusCount(Integer stockSurplusCount) {
+        this.stockSurplusCount = stockSurplusCount;
+    }
 }

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/model/res/StockResult.java

@@ -0,0 +1,47 @@
+package com.seamew.lottery.domain.activity.model.res;
+
+import com.seamew.lottery.common.Result;
+
+/**
+ * @Author: seamew
+ * @Title: StockResult
+ * @CreateTime: 2023年03月03日 10:03:00
+ * @Description: 库存处理结果
+ * @Version: 1.0
+ */
+public class StockResult extends Result {
+    /**
+     * 库存 Key
+     */
+    private String stockKey;
+    /**
+     * activity 库存剩余
+     */
+    private Integer stockSurplusCount;
+
+    public StockResult(String code, String info) {
+        super(code, info);
+    }
+
+    public StockResult(String code, String info, String stockKey, Integer stockSurplusCount) {
+        super(code, info);
+        this.stockKey = stockKey;
+        this.stockSurplusCount = stockSurplusCount;
+    }
+
+    public String getStockKey() {
+        return stockKey;
+    }
+
+    public void setStockKey(String stockKey) {
+        this.stockKey = stockKey;
+    }
+
+    public Integer getStockSurplusCount() {
+        return stockSurplusCount;
+    }
+
+    public void setStockSurplusCount(Integer stockSurplusCount) {
+        this.stockSurplusCount = stockSurplusCount;
+    }
+}

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/model/vo/ActivityBillVO.java

@@ -23,41 +23,45 @@ public class ActivityBillVO {
     private String uId;
 
     /**
-     * 活动ID
+     * activity 活动ID
      */
     private Long activityId;
     /**
-     * 活动名称
+     * activity 活动名称
      */
     private String activityName;
     /**
-     * 开始时间
+     * activity 开始时间
      */
     private Date beginDateTime;
     /**
-     * 结束时间
+     * activity 结束时间
      */
     private Date endDateTime;
     /**
-     * 库存剩余
+     * 库存
+     */
+    private Integer stockCount;
+    /**
+     * activity 库存剩余
      */
     private Integer stockSurplusCount;
     /**
-     * 活动状态：1编辑、2提审、3撤审、4通过、5运行(审核通过后worker扫描状态)、6拒绝、7关闭、8开启
+     * activity 活动状态：1编辑、2提审、3撤审、4通过、5运行(审核通过后worker扫描状态)、6拒绝、7关闭、8开启
      * Constants.ActivityState
      */
     private Integer state;
     /**
-     * 策略ID
+     * activity 策略ID
      */
     private Long strategyId;
-
     /**
-     * 每人可参与次数
+     * activity 每人可参与次数
      */
     private Integer takeCount;
+
     /**
-     * 已领取次数
+     * user_take_activity_count 已领取次数
      */
     private Integer userTakeLeftCount;
 }

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/model/vo/ActivityPartakeRecordVO.java

@@ -0,0 +1,35 @@
+package com.seamew.lottery.domain.activity.model.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @Author: seamew
+ * @Title: ActivityPartakeRecordVO
+ * @CreateTime: 2023年03月03日 16:32:00
+ * @Description: 活动参与记录
+ * @Version: 1.0
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class ActivityPartakeRecordVO {
+
+    /**
+     * 用户ID
+     */
+    private String uId;
+    /**
+     * activity 活动ID
+     */
+    private Long activityId;
+    /**
+     * 库存
+     */
+    private Integer stockCount;
+    /**
+     * activity 库存剩余
+     */
+    private Integer stockSurplusCount;
+}

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/repository/IActivityRepository.java

@@ -2,6 +2,7 @@ package com.seamew.lottery.domain.activity.repository;
 
 import com.seamew.lottery.common.Constants;
 import com.seamew.lottery.domain.activity.model.req.PartakeReq;
+import com.seamew.lottery.domain.activity.model.res.StockResult;
 import com.seamew.lottery.domain.activity.model.vo.*;
 
 import java.util.List;
@@ -72,4 +73,23 @@ public interface IActivityRepository {
      * @return 待处理的活动集合
      */
     List<ActivityVO> scanToDoActivityList(Long id);
+
+    /**
+     * 扣减活动库存，通过Redis
+     *
+     * @param uId        用户ID
+     * @param activityId 活动ID
+     * @param stockCount 总库存
+     * @return 扣减结果
+     */
+    StockResult subtractionActivityStockByRedis(String uId, Long activityId, Integer stockCount);
+
+    /**
+     * 恢复活动库存，通过Redis 【如果非常异常，则需要进行缓存库存恢复，只保证不超卖的特性，所以不保证一定能恢复占用库存，另外最终可以由任务进行补偿库存】
+     *
+     * @param activityId    活动ID
+     * @param tokenKey      分布式 KEY 用于清理
+     * @param code          状态
+     */
+    void recoverActivityCacheStockByRedis(Long activityId, String tokenKey, String code);
 }

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/repository/IUserTakeActivityRepository.java

@@ -1,5 +1,6 @@
 package com.seamew.lottery.domain.activity.repository;
 
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
 import com.seamew.lottery.domain.activity.model.vo.DrawOrderVO;
 import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
 import com.seamew.lottery.domain.activity.model.vo.UserTakeActivityVO;
@@ -85,4 +86,11 @@ public interface IUserTakeActivityRepository {
      * @return 发货单
      */
     List<InvoiceVO> scanInvoiceMqState();
+
+    /**
+     * 更新活动库存
+     *
+     * @param activityPartakeRecordVO   活动领取记录
+     */
+    void updateActivityStock(ActivityPartakeRecordVO activityPartakeRecordVO);
 }

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/service/partake/BaseActivityPartake.java

@@ -4,6 +4,7 @@ import com.seamew.lottery.common.Constants;
 import com.seamew.lottery.common.Result;
 import com.seamew.lottery.domain.activity.model.req.PartakeReq;
 import com.seamew.lottery.domain.activity.model.res.PartakeResult;
+import com.seamew.lottery.domain.activity.model.res.StockResult;
 import com.seamew.lottery.domain.activity.model.vo.ActivityBillVO;
 import com.seamew.lottery.domain.activity.model.vo.UserTakeActivityVO;
 import com.seamew.lottery.domain.support.ids.IIdGenerator;
@@ -29,7 +30,7 @@ public abstract class BaseActivityPartake extends ActivityPartakeSupport impleme
         // 1. 查询是否存在未执行抽奖领取活动单【user_take_activity 存在 state = 0，领取了但抽奖过程失败的，可以直接返回领取结果继续抽奖】
         UserTakeActivityVO userTakeActivityVO = queryNoConsumedTakeActivityOrder(req.getActivityId(), req.getUId());
         if (null != userTakeActivityVO) {
-            return buildPartakeResult(userTakeActivityVO.getStrategyId(), userTakeActivityVO.getTakeId());
+            return buildPartakeResult(userTakeActivityVO.getStrategyId(), userTakeActivityVO.getTakeId(), Constants.ResponseCode.NOT_CONSUMED_TAKE);
         }
 
         // 2. 查询活动账单
@@ -41,9 +42,11 @@ public abstract class BaseActivityPartake extends ActivityPartakeSupport impleme
             return new PartakeResult(checkResult.getCode(), checkResult.getInfo());
         }
 
-        // 4. 扣减活动库存【目前为直接对配置库中的 lottery.activity 直接操作表扣减库存，后续优化为Redis扣减】
-        Result subtractionActivityResult = subtractionActivityStock(req);
+        // 4. 扣减活动库存，通过Redis【活动库存扣减编号，作为锁的Key，缩小颗粒度】 Begin
+        StockResult subtractionActivityResult = subtractionActivityStockByRedis(req.getUId(), req.getActivityId(), activityBillVO.getStockCount());
+
         if (!Constants.ResponseCode.SUCCESS.getCode().equals(subtractionActivityResult.getCode())) {
+            recoverActivityCacheStockByRedis(req.getActivityId(), subtractionActivityResult.getStockKey(), subtractionActivityResult.getCode());
             return new PartakeResult(subtractionActivityResult.getCode(), subtractionActivityResult.getInfo());
         }
 
@@ -51,10 +54,33 @@ public abstract class BaseActivityPartake extends ActivityPartakeSupport impleme
         Long takeId = idGeneratorMap.get(Constants.Ids.SnowFlake).nextId();
         Result grabResult = grabActivity(req, activityBillVO, takeId);
         if (!Constants.ResponseCode.SUCCESS.getCode().equals(grabResult.getCode())) {
+            recoverActivityCacheStockByRedis(req.getActivityId(), subtractionActivityResult.getStockKey(), grabResult.getCode());
             return new PartakeResult(grabResult.getCode(), grabResult.getInfo());
         }
 
-        return buildPartakeResult(activityBillVO.getStrategyId(), takeId);
+        // 6. 扣减活动库存，通过Redis End
+        recoverActivityCacheStockByRedis(req.getActivityId(), subtractionActivityResult.getStockKey(), Constants.ResponseCode.SUCCESS.getCode());
+
+        return buildPartakeResult(activityBillVO.getStrategyId(), takeId, activityBillVO.getStockCount(), subtractionActivityResult.getStockSurplusCount(), Constants.ResponseCode.SUCCESS);
+    }
+
+    /**
+     * 封装结果【返回的策略ID，用于继续完成抽奖步骤】
+     *
+     * @param strategyId        策略ID
+     * @param takeId            领取ID
+     * @param stockCount        库存
+     * @param stockSurplusCount 剩余库存
+     * @param code              状态码
+     * @return 封装结果
+     */
+    private PartakeResult buildPartakeResult(Long strategyId, Long takeId, Integer stockCount, Integer stockSurplusCount, Constants.ResponseCode code) {
+        PartakeResult partakeResult = new PartakeResult(code.getCode(), code.getInfo());
+        partakeResult.setStrategyId(strategyId);
+        partakeResult.setTakeId(takeId);
+        partakeResult.setStockCount(stockCount);
+        partakeResult.setStockSurplusCount(stockSurplusCount);
+        return partakeResult;
     }
 
     /**
@@ -62,10 +88,11 @@ public abstract class BaseActivityPartake extends ActivityPartakeSupport impleme
      *
      * @param strategyId 策略ID
      * @param takeId     领取ID
+     * @param code       状态码
      * @return 封装结果
      */
-    private PartakeResult buildPartakeResult(Long strategyId, Long takeId) {
-        PartakeResult partakeResult = new PartakeResult(Constants.ResponseCode.SUCCESS.getCode(), Constants.ResponseCode.SUCCESS.getInfo());
+    private PartakeResult buildPartakeResult(Long strategyId, Long takeId, Constants.ResponseCode code) {
+        PartakeResult partakeResult = new PartakeResult(code.getCode(), code.getInfo());
         partakeResult.setStrategyId(strategyId);
         partakeResult.setTakeId(takeId);
         return partakeResult;
@@ -97,6 +124,25 @@ public abstract class BaseActivityPartake extends ActivityPartakeSupport impleme
      */
     protected abstract Result subtractionActivityStock(PartakeReq req);
 
+    /**
+     * 扣减活动库存，通过Redis
+     *
+     * @param uId        用户ID
+     * @param activityId 活动号
+     * @param stockCount 总库存
+     * @return 扣减结果
+     */
+    protected abstract StockResult subtractionActivityStockByRedis(String uId, Long activityId, Integer stockCount);
+
+    /**
+     * 恢复活动库存，通过Redis 【如果非常异常，则需要进行缓存库存恢复，只保证不超卖的特性，所以不保证一定能恢复占用库存，另外最终可以由任务进行补偿库存】
+     *
+     * @param activityId 活动ID
+     * @param tokenKey   分布式 KEY 用于清理
+     * @param code       状态
+     */
+    protected abstract void recoverActivityCacheStockByRedis(Long activityId, String tokenKey, String code);
+
     /**
      * 领取活动
      *

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/service/partake/IActivityPartake.java

@@ -3,6 +3,7 @@ package com.seamew.lottery.domain.activity.service.partake;
 import com.seamew.lottery.common.Result;
 import com.seamew.lottery.domain.activity.model.req.PartakeReq;
 import com.seamew.lottery.domain.activity.model.res.PartakeResult;
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
 import com.seamew.lottery.domain.activity.model.vo.DrawOrderVO;
 import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
 
@@ -58,4 +59,11 @@ public interface IActivityPartake {
      * @return 发货单
      */
     List<InvoiceVO> scanInvoiceMqState(int dbCount, int tbCount);
+
+    /**
+     * 更新活动库存
+     *
+     * @param activityPartakeRecordVO   活动领取记录
+     */
+    void updateActivityStock(ActivityPartakeRecordVO activityPartakeRecordVO);
 }

lottery-domain/src/main/java/com/seamew/lottery/domain/activity/service/partake/impl/ActivityPartakeImpl.java

@@ -3,10 +3,8 @@ package com.seamew.lottery.domain.activity.service.partake.impl;
 import com.seamew.lottery.common.Constants;
 import com.seamew.lottery.common.Result;
 import com.seamew.lottery.domain.activity.model.req.PartakeReq;
-import com.seamew.lottery.domain.activity.model.vo.ActivityBillVO;
-import com.seamew.lottery.domain.activity.model.vo.DrawOrderVO;
-import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
-import com.seamew.lottery.domain.activity.model.vo.UserTakeActivityVO;
+import com.seamew.lottery.domain.activity.model.res.StockResult;
+import com.seamew.lottery.domain.activity.model.vo.*;
 import com.seamew.lottery.domain.activity.repository.IUserTakeActivityRepository;
 import com.seamew.lottery.domain.activity.service.partake.BaseActivityPartake;
 import com.seamew.lottery.domain.support.ids.IIdGenerator;
@@ -84,6 +82,16 @@ public class ActivityPartakeImpl extends BaseActivityPartake {
         return Result.buildSuccessResult();
     }
 
+    @Override
+    protected StockResult subtractionActivityStockByRedis(String uId, Long activityId, Integer stockCount) {
+        return activityRepository.subtractionActivityStockByRedis(uId, activityId, stockCount);
+    }
+
+    @Override
+    protected void recoverActivityCacheStockByRedis(Long activityId, String tokenKey, String code) {
+        activityRepository.recoverActivityCacheStockByRedis(activityId, tokenKey, code);
+    }
+
     @Override
     protected Result grabActivity(PartakeReq partake, ActivityBillVO bill, Long takeId) {
         try {
@@ -163,4 +171,9 @@ public class ActivityPartakeImpl extends BaseActivityPartake {
             dbRouter.clear();
         }
     }
+
+    @Override
+    public void updateActivityStock(ActivityPartakeRecordVO activityPartakeRecordVO) {
+        userTakeActivityRepository.updateActivityStock(activityPartakeRecordVO);
+    }
 }

lottery-infrastructure/pom.xml

@@ -26,6 +26,10 @@
             <groupId>com.seamew</groupId>
             <artifactId>db-router-spring-boot-starter</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.seamew</groupId>
             <artifactId>lottery-domain</artifactId>

lottery-infrastructure/src/main/java/com/seamew/lottery/infrastructure/dao/IActivityDao.java

@@ -55,4 +55,11 @@ public interface IActivityDao {
      */
     List<Activity> scanToDoActivityList(Long id);
 
+    /**
+     * 更新用户领取活动后，活动库存
+     *
+     * @param activity  入参
+     */
+    void updateActivityStock(Activity activity);
+
 }

lottery-infrastructure/src/main/java/com/seamew/lottery/infrastructure/repository/ActivityRepository.java

@@ -2,10 +2,14 @@ package com.seamew.lottery.infrastructure.repository;
 
 import com.seamew.lottery.common.Constants;
 import com.seamew.lottery.domain.activity.model.req.PartakeReq;
+import com.seamew.lottery.domain.activity.model.res.StockResult;
 import com.seamew.lottery.domain.activity.model.vo.*;
 import com.seamew.lottery.domain.activity.repository.IActivityRepository;
 import com.seamew.lottery.infrastructure.dao.*;
 import com.seamew.lottery.infrastructure.po.*;
+import com.seamew.lottery.infrastructure.util.RedisUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Repository;
 
@@ -21,6 +25,7 @@ import java.util.List;
  * @Version: 1.0
  */
 @Repository
+@Slf4j
 public class ActivityRepository implements IActivityRepository {
 
     @Resource
@@ -33,13 +38,16 @@ public class ActivityRepository implements IActivityRepository {
     private IStrategyDetailDao strategyDetailDao;
     @Resource
     private IUserTakeActivityCountDao userTakeActivityCountDao;
-
+    @Resource
+    private RedisUtil redisUtil;
 
     @Override
     public void addActivity(ActivityVO activity) {
         Activity req = new Activity();
         BeanUtils.copyProperties(activity, req);
         activityDao.insert(req);
+        // 设置活动库存 KEY
+        redisUtil.set(Constants.RedisKey.KEY_LOTTERY_ACTIVITY_STOCK_COUNT(activity.getActivityId()), 0);
     }
 
     @Override
@@ -81,9 +89,15 @@ public class ActivityRepository implements IActivityRepository {
     @Override
     public ActivityBillVO queryActivityBill(PartakeReq req) {
 
+
         // 查询活动信息
         Activity activity = activityDao.queryActivityById(req.getActivityId());
 
+        // 从缓存中获取库存
+        String usedStockCountObj = redisUtil.get(Constants.RedisKey.KEY_LOTTERY_ACTIVITY_STOCK_COUNT(req.getActivityId()));
+        if (StringUtils.isEmpty(usedStockCountObj)) {
+            redisUtil.set(Constants.RedisKey.KEY_LOTTERY_ACTIVITY_STOCK_COUNT(req.getActivityId()), activity.getStockCount() - activity.getStockSurplusCount());
+        }
         // 查询领取次数
         UserTakeActivityCount userTakeActivityCountReq = new UserTakeActivityCount();
         userTakeActivityCountReq.setUId(req.getUId());
@@ -98,11 +112,11 @@ public class ActivityRepository implements IActivityRepository {
         activityBillVO.setBeginDateTime(activity.getBeginDateTime());
         activityBillVO.setEndDateTime(activity.getEndDateTime());
         activityBillVO.setTakeCount(activity.getTakeCount());
-        activityBillVO.setStockSurplusCount(activity.getStockSurplusCount());
+        activityBillVO.setStockCount(activity.getStockCount());
+        activityBillVO.setStockSurplusCount(null == usedStockCountObj ? activity.getStockSurplusCount() : activity.getStockCount() - Integer.parseInt(usedStockCountObj));
         activityBillVO.setStrategyId(activity.getStrategyId());
         activityBillVO.setState(activity.getState());
         activityBillVO.setUserTakeLeftCount(null == userTakeActivityCount ? null : userTakeActivityCount.getLeftCount());
-
         return activityBillVO;
     }
 
@@ -128,4 +142,38 @@ public class ActivityRepository implements IActivityRepository {
         return activityVOList;
     }
 
+    @Override
+    public StockResult subtractionActivityStockByRedis(String uId, Long activityId, Integer stockCount) {
+
+        //  1. 获取抽奖活动库存 Key
+        String stockKey = Constants.RedisKey.KEY_LOTTERY_ACTIVITY_STOCK_COUNT(activityId);
+
+        // 2. 扣减库存，目前占用库存数
+        Integer stockUsedCount = redisUtil.incr(stockKey, 1).intValue();
+
+        // 3. 超出库存判断，进行恢复原始库存
+        if (stockUsedCount > stockCount) {
+            redisUtil.decr(stockKey, 1);
+            return new StockResult(Constants.ResponseCode.OUT_OF_STOCK.getCode(), Constants.ResponseCode.OUT_OF_STOCK.getInfo());
+        }
+
+        // 4. 以活动库存占用编号，生成对应加锁Key，细化锁的颗粒度
+        String stockTokenKey = Constants.RedisKey.KEY_LOTTERY_ACTIVITY_STOCK_COUNT_TOKEN(activityId, stockUsedCount);
+
+        // 5. 使用 Redis.setNx 加一个分布式锁
+        boolean lockToken = redisUtil.setNx(stockTokenKey, 350L);
+        if (!lockToken) {
+            log.info("抽奖活动{}用户秒杀{}扣减库存，分布式锁失败：{}", activityId, uId, stockTokenKey);
+            return new StockResult(Constants.ResponseCode.ERR_TOKEN.getCode(), Constants.ResponseCode.ERR_TOKEN.getInfo());
+        }
+
+        return new StockResult(Constants.ResponseCode.SUCCESS.getCode(), Constants.ResponseCode.SUCCESS.getInfo(), stockTokenKey, stockCount - stockUsedCount);
+    }
+
+    @Override
+    public void recoverActivityCacheStockByRedis(Long activityId, String tokenKey, String code) {
+        // 删除分布式锁 Key
+        redisUtil.del(tokenKey);
+    }
+
 }

lottery-infrastructure/src/main/java/com/seamew/lottery/infrastructure/repository/UserTakeActivityRepository.java

@@ -1,13 +1,16 @@
 package com.seamew.lottery.infrastructure.repository;
 
 import com.seamew.lottery.common.Constants;
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
 import com.seamew.lottery.domain.activity.model.vo.DrawOrderVO;
 import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
 import com.seamew.lottery.domain.activity.model.vo.UserTakeActivityVO;
 import com.seamew.lottery.domain.activity.repository.IUserTakeActivityRepository;
+import com.seamew.lottery.infrastructure.dao.IActivityDao;
 import com.seamew.lottery.infrastructure.dao.IUserStrategyExportDao;
 import com.seamew.lottery.infrastructure.dao.IUserTakeActivityCountDao;
 import com.seamew.lottery.infrastructure.dao.IUserTakeActivityDao;
+import com.seamew.lottery.infrastructure.po.Activity;
 import com.seamew.lottery.infrastructure.po.UserStrategyExport;
 import com.seamew.lottery.infrastructure.po.UserTakeActivity;
 import com.seamew.lottery.infrastructure.po.UserTakeActivityCount;
@@ -28,6 +31,9 @@ import java.util.List;
 @Component
 public class UserTakeActivityRepository implements IUserTakeActivityRepository {
 
+    @Resource
+    private IActivityDao activityDao;
+
     @Resource
     private IUserTakeActivityCountDao userTakeActivityCountDao;
 
@@ -158,4 +164,12 @@ public class UserTakeActivityRepository implements IUserTakeActivityRepository {
         return invoiceVOList;
     }
 
+    @Override
+    public void updateActivityStock(ActivityPartakeRecordVO activityPartakeRecordVO) {
+        Activity activity = new Activity();
+        activity.setActivityId(activityPartakeRecordVO.getActivityId());
+        activity.setStockSurplusCount(activityPartakeRecordVO.getStockSurplusCount());
+        activityDao.updateActivityStock(activity);
+    }
+
 }

lottery-infrastructure/src/main/java/com/seamew/lottery/infrastructure/util/RedisUtil.java

@@ -0,0 +1,129 @@
+package com.seamew.lottery.infrastructure.util;
+
+import com.sun.org.apache.xpath.internal.operations.Bool;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.ObjectUtils;
+
+import javax.annotation.Resource;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Objects;
+
+/**
+ * @Author: seamew
+ * @Title: RedisUtil
+ * @CreateTime: 2023年03月03日 10:26:00
+ * @Description: Redis 工具类
+ * @Version: 1.0
+ */
+@Component
+public class RedisUtil {
+    @Autowired
+    private StringRedisTemplate redisTemplate;
+
+    /**
+     * 递增
+     *
+     * @param key   键
+     * @param delta 要增加几(大于0)
+     * @return 执行结果
+     */
+    public Long incr(String key, long delta) {
+        if (delta < 0) {
+            throw new RuntimeException("递增因子必须大于0");
+        }
+        return redisTemplate.opsForValue().increment(key, delta);
+    }
+
+    /**
+     * 递减
+     *
+     * @param key   键
+     * @param delta 要减少几(小于0)
+     * @return 执行结果
+     */
+    public Long decr(String key, long delta) {
+        if (delta < 0) {
+            throw new RuntimeException("递减因子必须大于0");
+        }
+        return redisTemplate.opsForValue().increment(key, -delta);
+    }
+
+    /**
+     * 分布式锁
+     *
+     * @param key            锁住的key
+     * @param lockExpireMils 锁住的时长。如果超时未解锁，视为加锁线程死亡，其他线程可夺取锁
+     * @return 执行结果
+     */
+    public Boolean setNx(String key, Long lockExpireMils) {
+        return (Boolean) redisTemplate.execute((RedisCallback<?>) connection -> {
+            // 获取时间毫秒值
+            long expireAt = System.currentTimeMillis() + lockExpireMils + 1;
+            //获取锁
+            Boolean acquire = connection.setNX(key.getBytes(), String.valueOf(expireAt).getBytes());
+            // 增加超时处理逻辑
+            if (Boolean.TRUE.equals(acquire)) {
+                return true;
+            }
+            byte[] value = connection.get(key.getBytes());
+
+            if (Objects.nonNull(value) && value.length > 0) {
+                // 将获取到的值转为long类型，便于进行比较
+                long expireTime = Long.parseLong(new String(value));
+                // 如果超时，重新加锁，防止死锁
+                if (expireTime < System.currentTimeMillis()) {
+                    byte[] oldValue = connection.getSet(key.getBytes(), String.valueOf(System.currentTimeMillis() + lockExpireMils + 1).getBytes());
+                    return Long.parseLong(new String(oldValue)) < System.currentTimeMillis();
+                }
+            }
+            return false;
+        });
+    }
+
+    /**
+     * 删除缓存
+     *
+     * @param key 一个或多个
+     */
+    public void del(String... key) {
+        if (key != null && key.length > 0) {
+            if (key.length == 1) {
+                redisTemplate.delete(key[0]);
+            } else {
+                redisTemplate.delete(Arrays.asList(key));
+            }
+        }
+    }
+
+    /**
+     * 普通缓存获取
+     *
+     * @param key 键
+     * @return 值
+     */
+    public String get(String key) {
+        return redisTemplate.opsForValue().get(key);
+    }
+
+    /**
+     * 普通缓存放入
+     *
+     * @param key   键
+     * @param value 值
+     * @return true成功 false失败
+     */
+    public boolean set(String key, Object value) {
+        try {
+            redisTemplate.opsForValue().set(key, String.valueOf(value));
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+}

lottery-interfaces/src/main/resources/application.yaml

@@ -40,6 +40,13 @@ spring:
       #listner负责ack，每调用一次，就立即commit
       ack-mode: manual_immediate
       missing-topics-fatal: false
+  # redis
+  redis:
+    database: 1
+    host: 180.76.231.231
+    port: 9736
+    password: sunhaobo
+    timeout: 5000
 
 # xxl-job
 # 官网：https://github.com/xuxueli/xxl-job/
@@ -49,7 +56,7 @@ spring:
 xxl:
   job:
     admin:
-      addresses: http://127.0.0.1:7397/xxl-job-admin
+      addresses: http://180.76.231.231:7397/xxl-job-admin
     executor:
       address:
       appname: lottery-job

lottery-interfaces/src/main/resources/mybatis/mapper/Activity_Mapper.xml

@@ -10,7 +10,8 @@
                 #{stockCount}, #{stockSurplusCount}, #{takeCount}, #{strategyId}, #{state}, #{creator}, now(), now())
     </insert>
 
-    <select id="queryActivityById" parameterType="java.lang.Long" resultType="com.seamew.lottery.infrastructure.po.Activity">
+    <select id="queryActivityById" parameterType="java.lang.Long"
+            resultType="com.seamew.lottery.infrastructure.po.Activity">
         SELECT activity_id,
                activity_name,
                activity_desc,
@@ -36,11 +37,14 @@
     </update>
 
     <update id="subtractionActivityStock" parameterType="java.lang.Long">
-        UPDATE activity SET stock_surplus_count = stock_surplus_count - 1
-        WHERE activity_id = #{activityId} AND stock_surplus_count > 0
+        UPDATE activity
+        SET stock_surplus_count = stock_surplus_count - 1
+        WHERE activity_id = #{activityId}
+          AND stock_surplus_count > 0
     </update>
 
-    <select id="scanToDoActivityList" parameterType="java.lang.Long" resultType="com.seamew.lottery.infrastructure.po.Activity">
+    <select id="scanToDoActivityList" parameterType="java.lang.Long"
+            resultType="com.seamew.lottery.infrastructure.po.Activity">
         SELECT activity_id, activity_name, begin_date_time, end_date_time, state, creator
         FROM activity
         WHERE id >= #{id}
@@ -48,5 +52,12 @@
         ORDER BY ID ASC LIMIT 10
     </select>
 
+    <update id="updateActivityStock" parameterType="com.seamew.lottery.infrastructure.po.Activity">
+        UPDATE activity
+        SET stock_surplus_count = #{stockSurplusCount}
+        WHERE activity_id = #{activityId}
+          AND stock_surplus_count > #{stockSurplusCount}
+    </update>
+
 
 </mapper>

lottery-interfaces/src/test/java/com/seamew/lottery/test/application/KafkaProducerTest.java

@@ -2,6 +2,7 @@ package com.seamew.lottery.test.application;
 
 import com.seamew.lottery.application.mq.producer.KafkaProducer;
 import com.seamew.lottery.common.Constants;
+import com.seamew.lottery.domain.activity.model.vo.ActivityPartakeRecordVO;
 import com.seamew.lottery.domain.activity.model.vo.InvoiceVO;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -53,6 +54,16 @@ public class KafkaProducerTest {
         }
     }
 
+    @Test
+    public void test_partake() throws InterruptedException {
+        ActivityPartakeRecordVO activityPartakeRecord = new ActivityPartakeRecordVO();
+        activityPartakeRecord.setActivityId(100001L);
+        activityPartakeRecord.setStockSurplusCount(78);
+        kafkaProducer.sendLotteryActivityPartakeRecord(activityPartakeRecord);
+        Thread.sleep(1000);
+
+    }
+
     @Test
     public void context() {
         Map<String, Object> configs = new HashMap<>();
@@ -71,7 +82,7 @@ public class KafkaProducerTest {
         KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(configs);
 
         List<String> topics = new ArrayList<>();
-        topics.add(KafkaProducer.TOPIC_NAME);
+        topics.add(KafkaProducer.TOPIC_INVOICE);
         //消费者订阅主题
         consumer.subscribe(topics);
         while (true) {

lottery-interfaces/src/test/java/com/seamew/lottery/test/infrastructure/RedisUtilTest.java

@@ -0,0 +1,43 @@
+package com.seamew.lottery.test.infrastructure;
+
+import com.seamew.lottery.domain.activity.model.req.PartakeReq;
+import com.seamew.lottery.domain.activity.service.partake.IActivityPartake;
+import com.seamew.lottery.infrastructure.util.RedisUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import javax.annotation.Resource;
+
+/**
+ * @Author: seamew
+ * @Title: RedisUtilTest
+ * @CreateTime: 2023年03月03日 10:50:00
+ * @Description: Redis 使用测试
+ * @Version: 1.0
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+@Slf4j
+public class RedisUtilTest {
+    @Resource
+    private IActivityPartake activityPartake;
+    @Resource
+    private RedisUtil redisUtil;
+
+    @Test
+    public void test_set() {
+        PartakeReq partakeReq = new PartakeReq("xiaofuge", 100001L);
+        activityPartake.doPartake(partakeReq);
+    }
+
+    @Test
+    public void test_get() {
+        // redisUtil.set("test", 0);
+        // String test = redisUtil.get("test");
+        // System.out.println(test);
+        redisUtil.setNx("key", 100000L);
+    }
+}