用户点击 → Controller 加锁(SET NX EX 300s)

    ↓ 锁已存在 → 返回 "操作频繁，请勿重复提交！"
    ↓ 加锁成功
Service.fullOutput() → 发送N条Kafka消息 → Redis记录 count=N
    ↓ 发送失败 → 立即释放锁
    ↓
Kafka Consumer 逐条处理
    ↓ 每条处理完 → DECR count
    ↓ count <= 0 → 删除锁 + 删除计数器
    ↓
用户可再次操作

src/main/java/zs/payment/controller/EweiShopGoodsController.java

@@ -8,6 +8,7 @@ import zs.payment.req.CommodityProcureReq;
 import zs.payment.req.GoodsSupplyReq;
 import zs.payment.resp.Result;
 import zs.payment.service.commodity.CommodityProcureService;
+import zs.payment.utils.RedisUtils;
 
 import javax.validation.Valid;
 
@@ -18,6 +19,13 @@ public class EweiShopGoodsController {
 
     @Autowired
     private CommodityProcureService commodityProcureService;
+    @Autowired
+    private RedisUtils redisUtils;
+
+    /** 防重复提交锁前缀 */
+    private static final String LOCK_PREFIX_FULL_OUTPUT = "lock:fullOutput:";
+    /** 锁过期时间（秒），防止死锁 */
+    private static final int LOCK_EXPIRE_SECONDS = 30;
 
     /**
      * 进货：把云商城（悦店管理）里的商品全进货到本地
@@ -34,14 +42,27 @@ public class EweiShopGoodsController {
     /**
      * 供货/停止供货：把本地商品推送到云商城，或停止供货
      * issupply=1: 供货  issupply=0: 停止供货
+     * 加入Redis分布式锁防止用户多次点击重复提交
      */
     @PostMapping("/fullOutput")
     public Result fullOutput(@Valid @RequestBody GoodsSupplyReq req) {
+        // 防重复提交：基于 uniacid + issupply 加分布式锁
+        String lockKey = LOCK_PREFIX_FULL_OUTPUT + req.getUniacid() + ":" + req.getIssupply();
+        boolean locked = redisUtils.setIfAbsent(lockKey, "1", LOCK_EXPIRE_SECONDS);
+        if (!locked) {
+            return Result.fail("操作频繁，请勿重复提交！");
+        }
         try {
             return commodityProcureService.fullOutput(req);
         } catch (ApiException e) {
+            // 发送Kafka失败时立即释放锁
+            redisUtils.del(lockKey);
             return Result.fail(e.getMsg());
+        } catch (Exception e) {
+            redisUtils.del(lockKey);
+            throw e;
         }
+        // 正常情况不释放锁，由Kafka消费者处理完所有任务后释放
     }
 
 }

src/main/java/zs/payment/messages/KafkaConsumer.java

@@ -427,10 +427,22 @@ public class KafkaConsumer {
         Integer cate = jsonObject.getIntValue("cate");
         Integer storeid = jsonObject.getIntValue("storeid");
 
+        String lockKey = "lock:fullOutput:" + uniacid + ":" + issupply;
         try {
             commodityProcureService.processSupplyItem(goodsId, issupply, cate, uniacid, storeid);
         } catch (Exception e) {
             log.error("供货/停止供货处理失败 goodsId={}, issupply={}", goodsId, issupply, e);
+        } finally {
+            // 每处理完一条消息，计数器递减，归零时释放锁
+            try {
+                long remaining = redisUtils.decr(lockKey + ":count");
+                if (remaining <= 0) {
+                    redisUtils.del(lockKey);
+                    redisUtils.del(lockKey + ":count");
+                }
+            } catch (Exception ex) {
+                log.warn("释放供货锁异常 lockKey={}", lockKey, ex);
+            }
         }
     }
 

src/main/java/zs/payment/service/commodity/impl/CommodityProcureServiceImpl.java

@@ -786,6 +786,15 @@ public class CommodityProcureServiceImpl implements CommodityProcureService {
             }
         }
 
+        // 记录待处理任务数量到Redis，用于消费者端判断全部完成后释放锁
+        String lockKey = "lock:fullOutput:" + uniacid + ":" + issupply;
+        if (submitCount > 0) {
+            redisUtils.set(lockKey + ":count", String.valueOf(submitCount));
+        } else {
+            // 无任务提交，立即释放锁
+            redisUtils.del(lockKey);
+        }
+
         log.info("供货/停止供货任务已提交，共{}个商品待处理，总查询{}个", submitCount, items.size());
         return Result.success("任务已提交，共" + submitCount + "个商品将异步处理");
     }
@@ -1240,6 +1249,7 @@ public class CommodityProcureServiceImpl implements CommodityProcureService {
         // 停止供货时取消推荐状态
         ImsEweiShopGoods update = new ImsEweiShopGoods();
         update.setId(item.getId());
+        update.setGdId(null);
         update.setIssupplyRecommend(0);
         goodsService.updateById(update);
     }

src/main/java/zs/payment/utils/RedisUtils.java

@@ -82,6 +82,14 @@ public class RedisUtils {
         return Redis.withCluster(ecoCluster, jedis -> jedis.del(key));
     }
 
+    /**
+     * 对key的值做原子递减操作（-1）
+     * @return 递减后的值
+     */
+    public Long decr(String key) {
+        return Redis.withCluster(ecoCluster, jedis -> jedis.decr(key));
+    }
+
     public void hset(String key, String field, String cont) {
         hset(key, field, cont, 0);
     }