添加分布式锁

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

@@ -27,7 +27,6 @@ public class YxSupplyProductController {
      */
     @PostMapping("/cursorList")
     public Result cursorList(@Valid @RequestBody ProductPageReq req) {
-
         return productSupplyService.cursorList(req);
     }
 
@@ -47,10 +46,11 @@ public class YxSupplyProductController {
 
     /**
      * 同步到自研系统上
+     * @param uniacid 商户ID，仅允许 uniacid=463 调用
      * @return
      */
     @GetMapping("/syncToQiyun")
-    public Result syncToQiyun() {
-        return productSupplyService.syncToQiyun();
+    public Result syncToQiyun(@RequestParam(required = false) Integer uniacid) {
+        return productSupplyService.syncToQiyun(uniacid);
     }
 }

src/main/java/zs/payment/enums/ProductSupplyRedis.java

@@ -4,7 +4,15 @@ public interface ProductSupplyRedis {
 
     // 芸信 供应链的token
     String SUPPLY_TOKEN = "supply:token";
-    //
+    // 已同步商品ID集合
     String SYNC_SUPPLY_ID = "sync:supply:ids";
+    // 选品库同步分布式锁（防重复同步）
+    String SYNC_QIYUN_LOCK = "sync:qiyun:lock";
+    // 选品库同步分布式锁的TTL（单位：秒，30分钟）
+    int SYNC_QIYUN_LOCK_TTL = 1800;
+
+    // 后续同步分页查询供应链商品
+    String SYNC_QIYUN_REMAIN= "sync.qiyun.remain";
+
 
 }

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

@@ -446,6 +446,7 @@ public class KafkaConsumer {
         JSONObject jsonObject = JSONObject.parseObject(message);
         int total = jsonObject.getIntValue("total");
         int pageSize = jsonObject.getIntValue("pageSize");
+        Integer uniacid = jsonObject.getInteger("uniacid");
         int totalPages = (total + pageSize - 1) / pageSize;
 
         Map<String, String> headers = supplyTokenService.buildAuthHeaders();
@@ -492,6 +493,11 @@ public class KafkaConsumer {
 
         executor.shutdown();
         log.info("syncQiyunRemain: 所有分页处理完成，总页数={}", totalPages);
+
+        // 释放分布式锁，同步任务完成
+        String redisSyncLock = ProductSupplyRedis.SYNC_QIYUN_LOCK + uniacid;
+        redisUtils.del(redisSyncLock);
+        log.info("syncQiyunRemain: 同步任务完成，已释放分布式锁，uniacid={}", uniacid);
     }
 
     /**

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

@@ -20,6 +20,7 @@ import zs.payment.entity.mysql.flymall.B2cGoodsImgInOnline;
 import zs.payment.entity.mysql.flymall.B2cGoodsSkuInOnline;
 import zs.payment.entity.mysql.flymall.B2cGoodsSpuInOnline;
 import zs.payment.entity.mysql.flymall.B2cShop;
+import zs.payment.enums.ProductSupplyRedis;
 import zs.payment.exception.ApiException;
 import zs.payment.mapper.primary.ImsEweiShopApplyMapper;
 import zs.payment.mapper.primary.ImsEweiShopGoodsMapper;
@@ -103,8 +104,7 @@ public class CommodityProcureServiceImpl implements CommodityProcureService {
     private static final BigDecimal COMMISSION_DEFAULT_RATE = new BigDecimal("0.01");
     private static final Integer JD_SHOP_ID = 11507;
 
-    @Value("${mall.host:}")
-    private String mallHost;
+
 
     @Override
     public Result procureGoods(CommodityProcureReq req) {
@@ -720,6 +720,14 @@ public class CommodityProcureServiceImpl implements CommodityProcureService {
         Integer issupply = req.getIssupply();
         Integer cate = req.getCate();
 
+        // 检查商品同步锁：若同步进行中，拒绝执行供货操作
+        String redisSyncLock = ProductSupplyRedis.SYNC_QIYUN_LOCK + uniacid;
+        Boolean syncLockExists = redisUtils.exists(redisSyncLock);
+        if (Boolean.TRUE.equals(syncLockExists)) {
+            log.warn("fullOutput: 商品同步中，拒绝供货操作");
+            return Result.fail("商品同步中，请同步完成后再执行供货操作");
+        }
+
         // 1. 查询在云商城中的店铺id（需要supplystatus=1）
         ImsEweiShopApply apply = applyMapper.selectOne(new LambdaQueryWrapper<ImsEweiShopApply>()
                 .eq(ImsEweiShopApply::getUniacid, uniacid)

src/main/java/zs/payment/service/supply/product/ProductSupplyService.java

@@ -35,7 +35,7 @@ public interface ProductSupplyService {
      * 将选品库里的商品，同步到自研系统中
      * @return
      */
-    Result syncToQiyun();
+    Result syncToQiyun(Integer uniacid);
 
     /**
      * 查询选品库指定分页

src/main/java/zs/payment/service/supply/product/impl/ProductSupplyServiceImpl.java

@@ -7,6 +7,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import zs.payment.enums.CallbackMessageType;
+import zs.payment.enums.ProductSupplyRedis;
 import zs.payment.messages.KafkaProducer;
 import zs.payment.req.ProductDetailReq;
 import zs.payment.req.ProductPageReq;
@@ -15,6 +16,7 @@ import zs.payment.resp.Result;
 import zs.payment.service.supply.product.ProductSupplyService;
 import zs.payment.service.supply.token.SupplyTokenService;
 import zs.payment.utils.HttpUtil;
+import zs.payment.utils.RedisUtils;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -33,6 +35,9 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
     @Autowired
     private SupplyTokenService supplyTokenService;
 
+    @Autowired
+    private RedisUtils redisUtils;
+
     @Override
     public Result detailList(ProductDetailReq req) {
         Map<String, String> headers = supplyTokenService.buildAuthHeaders();
@@ -103,10 +108,25 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
     }
 
     @Override
-    public Result syncToQiyun() {
+    public Result syncToQiyun(Integer uniacid) {
+        // 权限校验：仅允许 uniacid = 463 调用
+        if (uniacid == null) {
+            //初期-前端不传参数，因此默认为463（云悦生活）
+            uniacid = 463;
+        }
+        String redisSyncLock = ProductSupplyRedis.SYNC_QIYUN_LOCK + uniacid;
+
+        // 尝试加分布式锁，防止重复同步
+        boolean lockAcquired = redisUtils.setIfAbsent(redisSyncLock, "1", ProductSupplyRedis.SYNC_QIYUN_LOCK_TTL);
+        if (!lockAcquired) {
+            log.warn("syncToQiyun: 商品已经在同步中，请勿重复同步操作");
+            return Result.fail("商品已经在同步中，请勿重复同步操作");
+        }
+
         Map<String, String> headers = supplyTokenService.buildAuthHeaders();
         if (headers == null) {
             log.error("syncToQiyun: 获取token失败");
+            redisUtils.del(redisSyncLock); // 释放锁
             return Result.fail("获取token失败");
         }
 
@@ -116,6 +136,8 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
             // 查询第一页，获取总数并发送
             JSONObject firstPageData = fetchStoragePage(1, pageSize, headers);
             if (firstPageData == null) {
+                log.error("syncToQiyun: 调用选品库列表接口失败");
+                redisUtils.del(redisSyncLock); // 释放锁
                 return Result.fail("调用选品库列表接口失败");
             }
 
@@ -131,7 +153,8 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
                 JSONObject taskMsg = new JSONObject();
                 taskMsg.put("total", total);
                 taskMsg.put("pageSize", pageSize);
-                kafkaProducer.sendMessage("sync.qiyun.remain", taskMsg.toJSONString());
+                taskMsg.put("uniacid", uniacid);
+                kafkaProducer.sendMessage(ProductSupplyRedis.SYNC_QIYUN_REMAIN, taskMsg.toJSONString());
             }
 
             log.info("syncToQiyun: 同步任务已提交，总计{}条", total);
@@ -139,6 +162,7 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
 
         } catch (Exception e) {
             log.error("syncToQiyun: 执行异常", e);
+            redisUtils.del(redisSyncLock); // 释放锁
             return Result.fail("执行异常: " + e.getMessage());
         }
     }

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

@@ -45,6 +45,15 @@ public class RedisUtils {
         return Redis.withCluster(ecoCluster, jedis -> jedis.get(key));
     }
 
+    /**
+     * 判断key是否存在
+     * @param key 键
+     * @return 如果key存在返回true，否则返回false
+     */
+    public Boolean exists(String key) {
+        return Redis.withCluster(ecoCluster, jedis -> jedis.exists(key));
+    }
+
     public List<String> lRange(String key, long startIndex, long endIndex) {
         List<String> back = new ArrayList<>();
         if (key == null || key.isEmpty()) {