package com.kxs.transfer.api.controller; 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.KxsDtsLog; import com.kxs.transfer.api.model.table.DMLData; import com.kxs.transfer.api.service.KxsDtsLogService; import com.kxs.transfer.api.service.user.KxsUserAddressService; import com.kxs.transfer.api.service.user.KxsUserService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.RestController; import java.util.List; /** *

* DTS控制器 * 在此控制器里做数据分发 * 比如将数据分发到不同的数据源中 *

* * @author 没秃顶的码农 * @date 2024-01-25 */ @RestController @Slf4j @RequiredArgsConstructor public class DtsController { private final KxsDtsLogService kxsDtsLogService; private final KxsUserService kxsUserService; private final KxsUserAddressService kxsUserAddressService; @DtsMsgListener public void dtsListener(Long dataId, DMLData dmlData, DefaultUserRecord record){ //过滤无效数据 if("Users".equals(dmlData.getTableName())){ List 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; } } //去重数据 KxsDtsLog dtsLog = kxsDtsLogService.getOne(Wrappers.lambdaQuery().eq(KxsDtsLog::getDataId, dataId)); if(dtsLog != null){ log.info("dts的数据重复:{},dataID:{}", dmlData.getValidFieldDataMap(), dataId); record.commit(String.valueOf(record.getSourceTimestamp())); return; } //用户表 if("Users".equals(dmlData.getTableName())){ kxsUserService.changeUser(dmlData); } //用户地址表 if("UserAddress".equals(dmlData.getTableName())){ kxsUserAddressService.changeData(dmlData); } //密码操作 if("UserMoveInfo".equals(dmlData.getTableName())){ kxsUserService.changeUserPwd(dmlData); } //存储历史数据 KxsDtsLog kxsDtsLog = new KxsDtsLog(); kxsDtsLog.setDataId(dataId); kxsDtsLog.setContent(JSON.toJSONString(dmlData)); kxsDtsLog.setTableName(dmlData.getTableName()); kxsDtsLog.setOperation(dmlData.getOperation().toString()); kxsDtsLogService.save(kxsDtsLog); log.info("消费dts的数据表信息:{}", dmlData.getValidFieldDataMap()); record.commit(String.valueOf(record.getSourceTimestamp())); } }