优化同步供应链商品到自研系统的代码

src/main/java/zs/payment/dao/supply/StorageSyncFailLogDao.java

@@ -0,0 +1,17 @@
+package zs.payment.dao.supply;
+
+import org.springframework.stereotype.Repository;
+import zs.payment.dao.NewMongoDBDao;
+import zs.payment.entity.mongodb.supply.StorageSyncFailLog;
+
+/**
+ * 选品库同步异常日志 Dao（使用 spring.data.mongodb.uri 主数据源）
+ */
+@Repository
+public class StorageSyncFailLogDao extends NewMongoDBDao<StorageSyncFailLog> {
+
+    @Override
+    protected Class<StorageSyncFailLog> getEntityClass() {
+        return StorageSyncFailLog.class;
+    }
+}

src/main/java/zs/payment/entity/mongodb/supply/StorageSyncFailLog.java

@@ -0,0 +1,42 @@
+package zs.payment.entity.mongodb.supply;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.bson.types.ObjectId;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.data.mongodb.core.mapping.Field;
+
+/**
+ * 选品库同步分页查询异常日志
+ */
+@Setter
+@Getter
+@NoArgsConstructor
+@Document(collection = "storage_sync_fail_log")
+public class StorageSyncFailLog {
+
+    @Id
+    private ObjectId id;
+
+    /** 查询页码 */
+    @Field("page")
+    private int page;
+
+    /** 每页大小 */
+    @Field("page_size")
+    private int pageSize;
+
+    /** 总数据量 */
+    @Field("total")
+    private int total;
+
+    /** 异常信息 */
+    @Field("error_msg")
+    private String errorMsg;
+
+    /** 创建时间 */
+    @Field("create_time")
+    private long createTime;
+}

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

@@ -25,11 +25,15 @@ import zs.payment.service.imseweishopgoodsspec.ImsEweiShopGoodsSpecService;
 import zs.payment.service.imseweishopgoodsspecitem.ImsEweiShopGoodsSpecItemService;
 import zs.payment.service.supply.paysupply.PaySupplyService;
 import zs.payment.service.supply.product.ProductSupplyService;
+import zs.payment.service.supply.token.SupplyTokenService;
 import zs.payment.utils.RedisUtils;
 import zs.payment.utils.TimeUtils;
+import zs.payment.dao.supply.StorageSyncFailLogDao;
+import zs.payment.entity.mongodb.supply.StorageSyncFailLog;
 
 import java.math.BigDecimal;
 import java.util.*;
+import java.util.concurrent.*;
 
 
 @Slf4j
@@ -41,6 +45,8 @@ public class KafkaConsumer {
     private PaySupplyService paySupplyService;
     @Autowired
     private ProductSupplyService productSupplyService;
+    @Autowired
+    private SupplyTokenService supplyTokenService;
 
     @Autowired
     private RedisUtils redisUtils;
@@ -58,7 +64,8 @@ public class KafkaConsumer {
     private ImsEweiShopGoodsSpecItemService imsEweiShopGoodsSpecItemService;
     @Autowired
     private ImsEweiShopGoodsOptionService imsEweiShopGoodsOptionService;
-
+    @Autowired
+    private StorageSyncFailLogDao storageSyncFailLogDao;
 
     /**
      * 商品下架
@@ -427,4 +434,80 @@ public class KafkaConsumer {
         }
     }
 
+    /**
+     * 选品库同步剩余分页并发处理
+     * 消息格式：{"total":100000, "pageSize":50}
+     */
+    @KafkaListener(
+        topics = "sync.qiyun.remain",
+        groupId = "pay2-group"
+    )
+    public void syncQiyunRemain(String message) {
+        JSONObject jsonObject = JSONObject.parseObject(message);
+        int total = jsonObject.getIntValue("total");
+        int pageSize = jsonObject.getIntValue("pageSize");
+        int totalPages = (total + pageSize - 1) / pageSize;
+
+        Map<String, String> headers = supplyTokenService.buildAuthHeaders();
+        if (headers == null) {
+            log.error("syncQiyunRemain: 获取token失败，无法处理剩余分页");
+            return;
+        }
+
+        // 线程池并发查询剩余分页（从第2页开始）
+        int threadCount = Math.min(20, totalPages - 1);
+        ExecutorService executor = Executors.newFixedThreadPool(Math.max(1, threadCount));
+        List<Future<?>> futures = new ArrayList<>();
+
+        for (int page = 2; page <= totalPages; page++) {
+            final int p = page;
+            futures.add(executor.submit(() -> {
+                try {
+                    JSONObject pageData = productSupplyService.fetchStoragePage(p, pageSize, headers);
+                    if (pageData == null) {
+                        log.error("syncQiyunRemain: 查询第{}页失败，返回null", p);
+                        saveSyncFailLog(p, pageSize, total, "查询第" + p + "页返回null");
+                        return;
+                    }
+                    JSONArray array = pageData.getJSONArray("list");
+                    if (array != null && !array.isEmpty()) {
+                        productSupplyService.sendKafkaMessageWithRetry(array.toString());
+                        log.info("syncQiyunRemain: 成功发送第{}页数据，共{}条", p, array.size());
+                    }
+                } catch (Exception e) {
+                    log.error("syncQiyunRemain: 查询第{}页异常", p, e);
+                    saveSyncFailLog(p, pageSize, total, e.getMessage());
+                }
+            }));
+        }
+
+        // 等待所有任务完成
+        for (Future<?> future : futures) {
+            try {
+                future.get();
+            } catch (Exception e) {
+                log.error("syncQiyunRemain: 等待分页任务完成异常", e);
+            }
+        }
+
+        executor.shutdown();
+        log.info("syncQiyunRemain: 所有分页处理完成，总页数={}", totalPages);
+    }
+
+    /**
+     * 保存同步异常日志到 MongoDB
+     */
+    private void saveSyncFailLog(int page, int pageSize, int total, String errorMsg) {
+        try {
+            StorageSyncFailLog logEntity = new StorageSyncFailLog();
+            logEntity.setPage(page);
+            logEntity.setPageSize(pageSize);
+            logEntity.setTotal(total);
+            logEntity.setErrorMsg(errorMsg);
+            logEntity.setCreateTime(System.currentTimeMillis());
+            storageSyncFailLogDao.save(logEntity);
+        } catch (Exception e) {
+            log.error("saveSyncFailLog: 保存异常日志到MongoDB失败", e);
+        }
+    }
 }

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

@@ -1,10 +1,13 @@
 package zs.payment.service.supply.product;
 
+import com.alibaba.fastjson.JSONObject;
 import zs.payment.req.ProductDetailReq;
 import zs.payment.req.ProductPageReq;
 import zs.payment.req.storage.StoragePageReq;
 import zs.payment.resp.Result;
 
+import java.util.Map;
+
 public interface ProductSupplyService {
 
     /**
@@ -34,4 +37,13 @@ public interface ProductSupplyService {
      */
     Result syncToQiyun();
 
+    /**
+     * 查询选品库指定分页
+     */
+    JSONObject fetchStoragePage(int page, int pageSize, Map<String, String> headers);
+
+    /**
+     * 发送 Kafka 消息（带重试）
+     */
+    boolean sendKafkaMessageWithRetry(String message);
 }

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

@@ -110,60 +110,32 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
             return Result.fail("获取token失败");
         }
 
-        int page = 1;
         int pageSize = 50;
-        int total = 0;
-        boolean isFirstPage = true;
 
         try {
-            while (true) {
-                Map<String, Object> params = new HashMap<>();
-                params.put("page", page);
-                params.put("pageSize", pageSize);
-                //1-倒序 2-升序
-                params.put("create_sort", 1);
-
-                String resp = HttpUtil.restTemplatePost(
-                    prefixUrl + "/supplyapi/app/product/storage/getMyStorageIdsList",
-                    params,
-                    headers
-                );
-
-                JSONObject respJson = JSONObject.parseObject(resp);
-                if (respJson.getIntValue("code") != 0) {
-                    log.error("syncToQiyun: 调用选品库列表接口失败，page={}，msg={}", page, respJson.getString("msg"));
-                    return Result.fail("调用选品库列表接口失败: " + respJson.getString("msg"));
-                }
-
-                JSONObject data = respJson.getJSONObject("data");
-                JSONArray array = data.getJSONArray("list");
-
-                if (isFirstPage) {
-                    total = data.getIntValue("total");
-                    isFirstPage = false;
-                }
-
-                if (array != null && !array.isEmpty()) {
-                    if (!sendKafkaMessageWithRetry(array.toString())) {
-                        log.error("syncToQiyun: Kafka发送失败，page={}，已重试3次", page);
-                        return Result.fail("Kafka发送失败，第" + page + "页数据发送失败");
-                    }
-                    log.info("syncToQiyun: 成功发送第{}页数据，共{}条", page, array.size());
-                }
-
-                int currentPage = data.getIntValue("page");
-                int currentPageSize = data.getIntValue("pageSize");
-                int nextPage = currentPage + 1;
+            // 查询第一页，获取总数并发送
+            JSONObject firstPageData = fetchStoragePage(1, pageSize, headers);
+            if (firstPageData == null) {
+                return Result.fail("调用选品库列表接口失败");
+            }
 
-                if (nextPage * currentPageSize > total) {
-                    break;
-                }
+            int total = firstPageData.getIntValue("total");
+            JSONArray array = firstPageData.getJSONArray("list");
+            if (array != null && !array.isEmpty()) {
+                sendKafkaMessageWithRetry(array.toString());
+                log.info("syncToQiyun: 成功发送第1页数据，共{}条", array.size());
+            }
 
-                page = nextPage;
+            // 发送异步任务消息，由KafkaConsumer并发处理剩余分页
+            if (total > (array == null ? 0 : array.size())) {
+                JSONObject taskMsg = new JSONObject();
+                taskMsg.put("total", total);
+                taskMsg.put("pageSize", pageSize);
+                kafkaProducer.sendMessage("sync.qiyun.remain", taskMsg.toJSONString());
             }
 
-            log.info("syncToQiyun: 所有商品数据同步完成，总计{}条", total);
-            return Result.success(true);
+            log.info("syncToQiyun: 同步任务已提交，总计{}条", total);
+            return Result.success("同步任务已提交，共" + total + "条数据将异步处理");
 
         } catch (Exception e) {
             log.error("syncToQiyun: 执行异常", e);
@@ -171,7 +143,31 @@ public class ProductSupplyServiceImpl implements ProductSupplyService {
         }
     }
 
-    private boolean sendKafkaMessageWithRetry(String message) {
+    /**
+     * 查询选品库指定分页
+     */
+    public JSONObject fetchStoragePage(int page, int pageSize, Map<String, String> headers) {
+        Map<String, Object> params = new HashMap<>();
+        params.put("page", page);
+        params.put("pageSize", pageSize);
+        params.put("create_sort", 1);
+
+        String resp = HttpUtil.restTemplatePost(
+            prefixUrl + "/supplyapi/app/product/storage/getMyStorageIdsList",
+            params,
+            headers
+        );
+
+        JSONObject respJson = JSONObject.parseObject(resp);
+        if (respJson == null || respJson.getIntValue("code") != 0) {
+            log.error("fetchStoragePage: 调用选品库列表接口失败，page={}，msg={}", page,
+                respJson != null ? respJson.getString("msg") : "null");
+            return null;
+        }
+        return respJson.getJSONObject("data");
+    }
+
+    public boolean sendKafkaMessageWithRetry(String message) {
         int maxRetries = 3;
 
         for (int retryCount = 0; retryCount < maxRetries; retryCount++) {