同步数据新增定时任务模块，消费失败的数据重新消费

kxs-transfer/src/main/java/com/kxs/transfer/api/KxsTransferApplication.java

@@ -6,6 +6,7 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 /**
  * 数据迁移模块 写入
@@ -16,6 +17,7 @@ import org.springframework.scheduling.annotation.EnableAsync;
 @EnableAsync
 @EnableDiscoveryClient
 @SpringBootApplication
+@EnableScheduling
 public class KxsTransferApplication {
     public static void main(String[] args) {
         SpringApplication.run(KxsTransferApplication.class, args);

kxs-transfer/src/main/java/com/kxs/transfer/api/task/taskJob.java

@@ -0,0 +1,120 @@
+package com.kxs.transfer.api.task;
+
+import cn.hutool.core.bean.BeanUtil;
+import com.alibaba.fastjson.JSON;
+import com.aliyun.dts.subscribe.clients.record.OperationType;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.kxs.transfer.api.model.KxsDtsErrorLog;
+import com.kxs.transfer.api.model.table.DMLData;
+import com.kxs.transfer.api.service.KxsDtsErrorLogService;
+import com.kxs.transfer.api.service.KxsDtsLogService;
+import com.kxs.transfer.api.service.product.*;
+import com.kxs.transfer.api.service.product.impl.KxsMachineTrackService;
+import com.kxs.transfer.api.service.product.impl.KxsShopOrderService;
+import com.kxs.transfer.api.service.store.KxsMachineRecycleService;
+import com.kxs.transfer.api.service.store.KxsWarehouseService;
+import com.kxs.transfer.api.service.sys.KxsMorningService;
+import com.kxs.transfer.api.service.sys.KxsSchoolStudyService;
+import com.kxs.transfer.api.service.sys.KxsSysMsgService;
+import com.kxs.transfer.api.service.user.*;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * 任务作业
+ *
+ * @author 没秃顶的码农
+ * @date 2024-04-10
+ */
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class taskJob {
+
+    private final KxsDtsLogService kxsDtsLogService;
+    private final KxsDtsErrorLogService kxsDtsErrorLogService;
+
+    private final KxsUserService kxsUserService;
+    private final KxsUserPresetLogService kxsUserPresetLogService;
+    private final KxsUserAddressService kxsUserAddressService;
+    private final KxsUserAmountService kxsUserAmountService;
+    private final KxsUserAdvanceService kxsUserAdvanceService;
+    private final KxsUserAmountLogService kxsUserAmountLogService;
+    private final KxsUserWithdrawalService kxsUserWithdrawalService;
+    private final KxsLeaderService kxsLeaderService;
+    private final KxsPartnerService kxsPartnerService;
+    private final KxsMorningService kxsMorningService;
+    private final KxsSchoolStudyService kxsSchoolStudyService;
+    private final KxsSysMsgService kxsSysMsgService;
+    private final KxsBankChangeLogService kxsBankChangeLogService;
+    private final KxsUserByStageService kxsUserByStageService;
+    private final KxsLeaderAmountLogService kxsLeaderAmountLogService;
+
+    private final KxsMachineService kxsMachineService;
+
+    //产品模块
+    private final KxsShopGoodsService kxsShopGoodsService;
+    private final KxsMachinePledgeService kxsMachinePledgeService;
+    private final KxsMachineRatioService kxsMachineRatioService;
+    private final KxsTicketService kxsTicketService;
+    private final KxsTicketTransferService kxsTicketTransferService;
+    private final KxsMerchantService kxsMerchantService;
+    private final KxsMachineTransferService kxsMachineTransferService;
+    private final KxsMachineTrackService kxsMachineTrackService;
+    private final KxsShopOrderService kxsShopOrderService;
+
+    //仓库模块
+    private final KxsWarehouseService kxsWarehouseService;
+    private final KxsMachineRecycleService kxsMachineRecycleService;
+
+
+    /**
+     * 同步数据失败重新消费
+     */
+    @Scheduled(fixedDelay = 1000 * 60 % 5)
+    public void tradeRepeatEntry(){
+        List<KxsDtsErrorLog> list = kxsDtsErrorLogService.list(Wrappers.<KxsDtsErrorLog>lambdaQuery().eq(KxsDtsErrorLog::getOperation, OperationType.INSERT).last("limit 100"));
+
+        for (KxsDtsErrorLog kxsDtsErrorLog : list) {
+            try {
+                //仓库担保记录
+                if ("StoreHouseAmountPromiss".equals(kxsDtsErrorLog.getTableName())) {
+                    kxsWarehouseService.changeCreditData(JSON.parseObject(kxsDtsErrorLog.getContent(), DMLData.class));
+                }
+                //商品订单同步
+                if ("Orders".equals(kxsDtsErrorLog.getTableName())) {
+                    kxsShopOrderService.changeData(JSON.parseObject(kxsDtsErrorLog.getContent(), DMLData.class));
+                }
+                //商品订单详情
+                if ("OrderProduct".equals(kxsDtsErrorLog.getTableName())) {
+                    kxsShopOrderService.changeInfoData(JSON.parseObject(kxsDtsErrorLog.getContent(), DMLData.class));
+                }
+
+
+
+
+
+
+
+
+                //删除本条记录
+                kxsDtsErrorLogService.removeById(kxsDtsErrorLog.getId());
+            } catch (Exception e) {
+                log.error("重复消费新增数据的队列消费失败",e);
+            }
+
+
+
+
+
+        }
+
+
+    }
+
+}