保存消费错误数据及错误信息

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

@@ -16,7 +16,6 @@ import org.springframework.scheduling.annotation.EnableAsync;
 @EnableAsync
 @EnableDiscoveryClient
 @SpringBootApplication
-@EnableAuthResourceServer
 public class KxsTransferApplication {
     public static void main(String[] args) {
         SpringApplication.run(KxsTransferApplication.class, args);

kxs-transfer/src/main/java/com/kxs/transfer/api/config/UserMetaStore.java

@@ -1,5 +1,9 @@
 package com.kxs.transfer.api.config;
 
+import cn.hutool.core.date.LocalDateTimeUtil;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.aliyun.dts.subscribe.clients.metastore.AbstractUserMetaStore;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.kxs.transfer.api.model.KxsCheckpoint;
@@ -20,19 +24,26 @@ public class UserMetaStore extends AbstractUserMetaStore {
     }
 
     @Override
-    protected void saveData(String groupID, String toStoreJson) {
+    protected void saveData(String groupId, String toStoreJson) {
+
+        JSONObject jsonObject = JSON.parseObject(toStoreJson);
+
+        JSONArray streamCheckpoint = jsonObject.getJSONArray("streamCheckpoint");
+        String info = streamCheckpoint.get(0).toString();
+        JSONObject infoObj = JSON.parseObject(info);
 
-        System.out.println("消费节点" + groupID+ "保存checkpoint数据：" + toStoreJson);
         KxsCheckpoint kxsCheckpoint = new KxsCheckpoint();
-        kxsCheckpoint.setGroupId(groupID);
+        kxsCheckpoint.setGroupId(groupId);
         kxsCheckpoint.setCheckpoint(toStoreJson);
+        if(infoObj != null && infoObj.getLong("timestamp") > 0){
+            kxsCheckpoint.setUpdateTime(LocalDateTimeUtil.of(infoObj.getLong("timestamp") * 1000));
+        }
         kxsCheckpointService.save(kxsCheckpoint);
     }
 
     @Override
     protected String getData(String groupId) {
 
-        System.out.println("获取节点：" + groupId);
         KxsCheckpoint checkpoint = kxsCheckpointService.getOne(Wrappers.lambdaQuery(KxsCheckpoint.class)
                 .eq(KxsCheckpoint::getGroupId, groupId)
                 .orderByDesc(KxsCheckpoint::getCreateTime)

kxs-transfer/src/main/java/com/kxs/transfer/api/controller/DtsController.java

@@ -4,8 +4,10 @@ import com.alibaba.fastjson.JSON;
 import com.aliyun.dts.subscribe.clients.record.DefaultUserRecord;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.kxs.transfer.api.annotation.DtsMsgListener;
+import com.kxs.transfer.api.model.KxsDtsErrorLog;
 import com.kxs.transfer.api.model.KxsDtsLog;
 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.user.KxsUserAddressService;
 import com.kxs.transfer.api.service.user.KxsUserAmountLogService;
@@ -13,15 +15,16 @@ import com.kxs.transfer.api.service.user.KxsUserAmountService;
 import com.kxs.transfer.api.service.user.KxsUserService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.exception.ExceptionUtils;
 import org.springframework.web.bind.annotation.RestController;
 
 import java.util.List;
 
 /**
  * <p>
- *     DTS控制器
- *     在此控制器里做数据分发
- *     比如将数据分发到不同的数据源中
+ * DTS控制器
+ * 在此控制器里做数据分发
+ * 比如将数据分发到不同的数据源中
  * </p>
  *
  * @author 没秃顶的码农
@@ -33,6 +36,7 @@ import java.util.List;
 public class DtsController {
 
     private final KxsDtsLogService kxsDtsLogService;
+    private final KxsDtsErrorLogService kxsDtsErrorLogService;
 
     private final KxsUserService kxsUserService;
     private final KxsUserAddressService kxsUserAddressService;
@@ -40,66 +44,86 @@ public class DtsController {
     private final KxsUserAmountLogService kxsUserAmountLogService;
 
     @DtsMsgListener
-    public void dtsListener(Long dataId, DMLData dmlData, DefaultUserRecord record){
+    public void dtsListener(Long dataId, DMLData dmlData, DefaultUserRecord record) {
 
-        //过滤无效数据
-        if("Users".equals(dmlData.getTableName())){
-            List<String> changeFieldList = dmlData.getChangeFieldList();
-            //过滤Users老表记录登陆设备的信息日志
-            if (changeFieldList.size() == 2 && changeFieldList.contains("DeviceId") && changeFieldList.contains("DeviceType")){
-                record.commit(String.valueOf(record.getSourceTimestamp()));
-                return;
+        try {
+            //过滤无效数据
+            if ("Users".equals(dmlData.getTableName())) {
+                List<String> changeFieldList = dmlData.getChangeFieldList();
+                //过滤Users老表记录登陆设备的信息日志
+                if (changeFieldList.size() == 2 && changeFieldList.contains("DeviceId") && changeFieldList.contains("DeviceType")) {
+                    record.commit(String.valueOf(record.getSourceTimestamp()));
+                    return;
+                }
+                if (changeFieldList.size() == 1 && changeFieldList.contains("DeviceId")) {
+                    record.commit(String.valueOf(record.getSourceTimestamp()));
+                    return;
+                }
+                //过滤Users本月交易额字段
+                if (changeFieldList.size() == 1 && changeFieldList.contains("ThisMonthTrade")) {
+                    record.commit(String.valueOf(record.getSourceTimestamp()));
+                    return;
+                }
             }
-            if (changeFieldList.size() == 1 && changeFieldList.contains("DeviceId")){
+
+            //去重数据
+            KxsDtsLog dtsLog = kxsDtsLogService.getOne(Wrappers.<KxsDtsLog>lambdaQuery().eq(KxsDtsLog::getDataId, dataId));
+            if (dtsLog != null) {
+                log.info("dts的数据重复:{}，dataID:{}", dmlData.getValidFieldDataMap(), dataId);
                 record.commit(String.valueOf(record.getSourceTimestamp()));
                 return;
             }
-            //过滤Users本月交易额字段
-            if (changeFieldList.size() == 1 && changeFieldList.contains("ThisMonthTrade")){
-                record.commit(String.valueOf(record.getSourceTimestamp()));
-                return;
+            log.info("开始消费:{}表的数据，dataID:{}", dmlData.getTableName(), dataId);
+
+            //用户表
+            if ("Users".equals(dmlData.getTableName())) {
+                kxsUserService.changeUser(dmlData);
+            }
+            //密码操作
+            if ("UserMoveInfo".equals(dmlData.getTableName())) {
+                kxsUserService.changeUserPwd(dmlData);
+            }
+            //用户地址表
+            if ("UserAddress".equals(dmlData.getTableName())) {
+                kxsUserAddressService.changeData(dmlData);
+            }
+            //用户账户
+            if ("UserAccount".equals(dmlData.getTableName())) {
+                kxsUserAmountService.changeData(dmlData);
+            }
+            //用户账户余额日志
+            if ("UserAccountRecord".equals(dmlData.getTableName())) {
+                kxsUserAmountLogService.changeData(dmlData);
             }
-        }
 
-        //去重数据
-        KxsDtsLog dtsLog = kxsDtsLogService.getOne(Wrappers.<KxsDtsLog>lambdaQuery().eq(KxsDtsLog::getDataId, dataId));
-        if(dtsLog != null){
-            log.info("dts的数据重复:{}，dataID:{}", dmlData.getValidFieldDataMap(), dataId);
-            record.commit(String.valueOf(record.getSourceTimestamp()));
-            return;
-        }
-        log.info("开始消费:{}表的数据，dataID:{}", dmlData.getTableName(), dataId);
+            //存储历史数据
+            KxsDtsLog kxsDtsLog = new KxsDtsLog();
+            kxsDtsLog.setDataId(dataId);
+            kxsDtsLog.setContent(JSON.toJSONString(dmlData));
+            kxsDtsLog.setTableName(dmlData.getTableName());
+            kxsDtsLog.setOperation(dmlData.getOperation().toString());
+            kxsDtsLogService.save(kxsDtsLog);
 
-        //用户表
-        if("Users".equals(dmlData.getTableName())){
-            kxsUserService.changeUser(dmlData);
-        }
-        //密码操作
-        if("UserMoveInfo".equals(dmlData.getTableName())){
-            kxsUserService.changeUserPwd(dmlData);
-        }
-        //用户地址表
-        if("UserAddress".equals(dmlData.getTableName())){
-            kxsUserAddressService.changeData(dmlData);
-        }
-        //用户账户
-        if("UserAccount".equals(dmlData.getTableName())){
-            kxsUserAmountService.changeData(dmlData);
-        }
-        //用户账户余额日志
-        if("UserAccountRecord".equals(dmlData.getTableName())){
-            kxsUserAmountLogService.changeData(dmlData);
-        }
+            //删除错误记录，不管有没有都删除，因为很快
+            kxsDtsErrorLogService.removeById(dataId);
+
+            record.commit(String.valueOf(record.getSourceTimestamp()));
+        } catch (Exception e) {
+            //详细错误信息
+            String fullStackTrace = ExceptionUtils.getFullStackTrace(e);
 
-        //存储历史数据
-        KxsDtsLog kxsDtsLog = new KxsDtsLog();
-        kxsDtsLog.setDataId(dataId);
-        kxsDtsLog.setContent(JSON.toJSONString(dmlData));
-        kxsDtsLog.setTableName(dmlData.getTableName());
-        kxsDtsLog.setOperation(dmlData.getOperation().toString());
-        kxsDtsLogService.save(kxsDtsLog);
+            //存储错误数据
+            KxsDtsErrorLog kxsDtsLog = new KxsDtsErrorLog();
+            kxsDtsLog.setId(dataId);
+            kxsDtsLog.setContent(JSON.toJSONString(dmlData));
+            kxsDtsLog.setTableName(dmlData.getTableName());
+            kxsDtsLog.setOperation(dmlData.getOperation().toString());
+            kxsDtsLog.setErrorStr(fullStackTrace);
+            kxsDtsErrorLogService.save(kxsDtsLog);
 
-        record.commit(String.valueOf(record.getSourceTimestamp()));
+            record.commit(String.valueOf(record.getSourceTimestamp()));
+            log.error(fullStackTrace);
+        }
     }
 
 }

kxs-transfer/src/main/java/com/kxs/transfer/api/mapper/KxsDtsErrorLogMapper.java

@@ -0,0 +1,19 @@
+package com.kxs.transfer.api.mapper;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.kxs.transfer.api.model.KxsDtsErrorLog;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据同步记录日志表(KxsDtsLog)表数据库访问层
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+@Mapper
+@DS("sys")
+public interface KxsDtsErrorLogMapper extends BaseMapper<KxsDtsErrorLog> {
+
+}
+

kxs-transfer/src/main/java/com/kxs/transfer/api/model/KxsCheckpoint.java

@@ -57,6 +57,12 @@ public class KxsCheckpoint extends Model<KxsCheckpoint> implements Serializable
     @Schema(description = "新增时间")
     private LocalDateTime createTime;
 
+    /**
+     * 消费节点时间
+     */
+    @Schema(description = "消费节点时间")
+    private LocalDateTime updateTime;
+
 
 
 

kxs-transfer/src/main/java/com/kxs/transfer/api/model/KxsDtsErrorLog.java

@@ -0,0 +1,83 @@
+package com.kxs.transfer.api.model;
+
+import com.baomidou.mybatisplus.annotation.FieldFill;
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * 数据同步记录失败日志表(KxsDtsErrorLog)表实体类
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+public class KxsDtsErrorLog extends Model<KxsDtsErrorLog> implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    
+    
+    /**
+     * 主键 数据ID
+     */
+    @Schema(description = "主键")
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+
+    /**
+     * 原始数据
+     */
+    @Schema(description = "原始数据")
+    private String content;
+
+    /**
+     * 操作类型
+     */
+    @Schema(description = "操作类型")
+    private String operation;
+
+    /**
+     * 表名
+     */
+    @Schema(description = "表名")
+    private String tableName;
+
+    /**
+     * 错误信息
+     */
+    @Schema(description = "错误信息")
+    private String errorStr;
+
+
+    /**
+     * 新增时间
+     */
+    @TableField(fill = FieldFill.INSERT)
+    @Schema(description = "新增时间")
+    private LocalDateTime createTime;
+
+
+    /**
+     * 修改时间
+     */
+    @TableField(fill = FieldFill.UPDATE)
+    @Schema(description = "修改时间")
+    private LocalDateTime updateTime;
+
+
+
+
+}
+

kxs-transfer/src/main/java/com/kxs/transfer/api/service/KxsDtsErrorLogService.java

@@ -0,0 +1,15 @@
+package com.kxs.transfer.api.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.kxs.transfer.api.model.KxsDtsErrorLog;
+
+/**
+ * 数据同步记录日志表(KxsDtsLog)表服务接口
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+public interface KxsDtsErrorLogService extends IService<KxsDtsErrorLog> {
+
+}
+

kxs-transfer/src/main/java/com/kxs/transfer/api/service/impl/KxsDtsErrorLogServiceImpl.java

@@ -0,0 +1,19 @@
+package com.kxs.transfer.api.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.kxs.transfer.api.mapper.KxsDtsErrorLogMapper;
+import com.kxs.transfer.api.model.KxsDtsErrorLog;
+import com.kxs.transfer.api.service.KxsDtsErrorLogService;
+import org.springframework.stereotype.Service;
+
+/**
+ * 数据同步记录日志表(KxsDtsLog)表服务实现类
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+@Service("kxsDtsErrorLogService")
+public class KxsDtsErrorLogServiceImpl extends ServiceImpl<KxsDtsErrorLogMapper, KxsDtsErrorLog> implements KxsDtsErrorLogService {
+
+}
+

kxs-transfer/src/main/java/com/kxs/transfer/api/service/user/impl/KxsUserAddressServiceImpl.java

@@ -10,6 +10,7 @@ import com.kxs.transfer.api.model.table.FieldData;
 import com.kxs.transfer.api.service.user.KxsUserAddressService;
 import com.kxs.user.api.model.KxsUser;
 import com.kxs.user.api.model.KxsUserAddress;
+import com.kxs.user.api.model.KxsUserAmount;
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
@@ -40,6 +41,11 @@ public class KxsUserAddressServiceImpl extends ServiceImpl<KxsUserAddressMapper,
         address.setId(id);
         //操作类型
         OperationType operation = dmlData.getOperation();
+        if(operation.equals(OperationType.DELETE)){
+            // 更新数据
+            baseMapper.deleteById(id);
+            return;
+        }
         for (String field : changeFieldList) {
 
             switch (field) {
@@ -80,8 +86,14 @@ public class KxsUserAddressServiceImpl extends ServiceImpl<KxsUserAddressMapper,
             baseMapper.updateById(address);
         }
         if(operation.equals(OperationType.INSERT)){
-            // 插入数据
-            baseMapper.insert(address);
+
+            KxsUserAddress kxsUserAddress = baseMapper.selectById(address.getId());
+            if (kxsUserAddress != null){
+                baseMapper.updateById(address);
+            }else{
+                // 插入数据
+                baseMapper.insert(address);
+            }
         }
     }
 }

kxs-transfer/src/main/java/com/kxs/transfer/api/service/user/impl/KxsUserAmountServiceImpl.java

@@ -87,8 +87,13 @@ public class KxsUserAmountServiceImpl extends ServiceImpl<KxsUserAmountMapper, K
             baseMapper.updateById(userAmount);
         }
         if(operation.equals(OperationType.INSERT)){
-            // 插入数据
-            baseMapper.insert(userAmount);
+            KxsUserAmount kxsUserAmount = baseMapper.selectById(userAmount.getId());
+            if (kxsUserAmount != null){
+                baseMapper.updateById(userAmount);
+            }else{
+                // 插入数据
+                baseMapper.insert(userAmount);
+            }
         }
     }
 }