初始化消费表数据

kxs-transfer/pom.xml

@@ -19,8 +19,20 @@
 
     <dependencies>
         <dependency>
-            <groupId>org.springframework.kafka</groupId>
-            <artifactId>spring-kafka</artifactId>
+            <groupId>com.aliyun.dts</groupId>
+            <artifactId>dts-new-subscribe-sdk</artifactId>
+            <version>2.0.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>2.7.0</version>
         </dependency>
 
         <!--注册中心客户端-->
@@ -59,6 +71,11 @@
             <groupId>com.kxs</groupId>
             <artifactId>kxs-common-core</artifactId>
         </dependency>
+        <!--安全模块-->
+        <dependency>
+            <groupId>com.kxs</groupId>
+            <artifactId>kxs-common-security</artifactId>
+        </dependency>
         <!-- orm 模块-->
         <dependency>
             <groupId>com.baomidou</groupId>

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

@@ -1,18 +1,24 @@
-//package com.kxs.transfer.api;
-//
-//
-//import org.springframework.boot.SpringApplication;
-//import org.springframework.boot.autoconfigure.SpringBootApplication;
-//
-///**
-// * 数据迁移模块 写入
-// *
-// * @author 没秃顶的码农
-// * @date 2024-01-19
-// */
-//@SpringBootApplication
-//public class KxsTransferApplication {
-//    public static void main(String[] args) {
-//        SpringApplication.run(KxsTransferApplication.class, args);
-//    }
-//}
+package com.kxs.transfer.api;
+
+
+import com.kxs.common.security.annotation.EnableAuthResourceServer;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+import org.springframework.scheduling.annotation.EnableAsync;
+
+/**
+ * 数据迁移模块 写入
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-19
+ */
+@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/annotation/DtsListener.java

@@ -0,0 +1,16 @@
+package com.kxs.transfer.api.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * DTS 侦听器
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-24
+ */
+@Target({ElementType.TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface DtsListener {
+
+}

kxs-transfer/src/main/java/com/kxs/transfer/api/annotation/DtsMsgListener.java

@@ -0,0 +1,16 @@
+package com.kxs.transfer.api.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * DTS 侦听器
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-24
+ */
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface DtsMsgListener {
+
+}

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

@@ -1,91 +1,95 @@
-//package com.kxs.transfer.api.config;
-//
-//import com.aliyun.dts.subscribe.clients.ConsumerContext;
-//import com.aliyun.dts.subscribe.clients.DTSConsumer;
-//import com.aliyun.dts.subscribe.clients.DefaultDTSConsumer;
-//import com.aliyun.dts.subscribe.clients.common.RecordListener;
-//import com.aliyun.dts.subscribe.clients.record.OperationType;
-//import com.aliyun.dts.subscribe.clients.recordprocessor.DbType;
-//import com.aliyun.dts.subscribe.clients.recordprocessor.DefaultRecordPrintListener;
-//import com.kxs.transfer.api.service.KxsCheckpointService;
-//import lombok.RequiredArgsConstructor;
-//import org.springframework.context.annotation.Bean;
-//import org.springframework.context.annotation.Configuration;
-//
-//import java.util.Collections;
-//import java.util.Map;
-//
-///**
-// * 分布式 DTS 订阅管道配置
-// *
-// * @author 没秃顶的码农
-// * @date 2024-01-19
-// */
-//@Configuration(proxyBeanMethods = false)
-//@RequiredArgsConstructor
-//public class ConsumerAssignConfig {
-//
-//    private final KxsCheckpointService kxsCheckpointService;
-//
-//    // kafka broker url
-//    String brokerUrl = "your broker url";
-//    // topic to consume, partition is 0
-//    String topic = "your dts topic";
-//    // user password and sid for auth
-//    String sid = "your sid";
-//    String userName = "your user name";
-//    String password = "your password";
-//    // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
-//    String initCheckpoint = "start timestamp";
-//    // when use subscribe mode, group config is required. kafka consumer group is enabled
-//    ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.ASSIGN;
-//    // if force use config checkpoint when start. for checkpoint reset, only assign mode works
-//    boolean isForceUseInitCheckpoint = true;
-//
-//    /**
-//     * 初始化 初始化消费者
-//     *
-//     * @return {@link DTSConsumer}
-//     */
-//    @Bean
-//    public DTSConsumer initDTSClient() {
-//        ConsumerContext consumerContext = new ConsumerContext(brokerUrl, topic, sid, userName, password, initCheckpoint, subscribeMode);
-//
-//        //if this parameter is set, force to use the initCheckpoint to initial
-//        consumerContext.setForceUseCheckpoint(isForceUseInitCheckpoint);
-//
-//        //add user store
-//        consumerContext.setUserRegisteredStore(new UserMetaStore(kxsCheckpointService));
-//
-//        DTSConsumer dtsConsumer = new DefaultDTSConsumer(consumerContext);
-//
-//        dtsConsumer.addRecordListeners(buildRecordListener());
-//        dtsConsumer.start();
-//        return dtsConsumer;
-//    }
-//
-//    public static Map<String, RecordListener> buildRecordListener() {
-//        // user can impl their own listener
-//        RecordListener mysqlRecordPrintListener = record -> {
-//
-//            OperationType operationType = record.getOperationType();
-//
-//            if(operationType.equals(OperationType.INSERT)
-//                    || operationType.equals(OperationType.UPDATE)
-//                    || operationType.equals(OperationType.DELETE)
-//                    || operationType.equals(OperationType.DDL)
-//                    || operationType.equals(OperationType.HEARTBEAT)) {
-//
-//                // consume record
-//                RecordListener recordPrintListener = new DefaultRecordPrintListener(DbType.MySQL);
-//
-//                recordPrintListener.consume(record);
-//
-//                //commit method push the checkpoint update
-//                record.commit("");
-//            }
-//        };
-//        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
-//    }
-//
-//}
+package com.kxs.transfer.api.config;
+
+import com.aliyun.dts.subscribe.clients.ConsumerContext;
+import com.aliyun.dts.subscribe.clients.DTSConsumer;
+import com.aliyun.dts.subscribe.clients.DefaultDTSConsumer;
+import com.aliyun.dts.subscribe.clients.common.RecordListener;
+import com.kxs.transfer.api.annotation.DtsListener;
+import com.kxs.transfer.api.service.KxsCheckpointService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.scheduling.annotation.Async;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *  DTS 订阅管道配置
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-19
+ */
+@Configuration(proxyBeanMethods = false)
+@RequiredArgsConstructor
+@EnableConfigurationProperties(DTSProperties.class)
+public class ConsumerAssignConfig implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {
+
+    private final KxsCheckpointService kxsCheckpointService;
+
+    private final DTSProperties dtsProperties;
+
+    DTSConsumer dtsConsumer = null;
+
+    @Override
+    @Async
+    public void onApplicationEvent(ContextRefreshedEvent event) {
+        System.out.println("异步开启 kafka客户端处理数据...");
+        //如果是第二次加载则不处理了
+        if(event.getApplicationContext().getParent() != null) {
+            return;
+        }
+        if(dtsConsumer != null){
+            return;
+        }
+        //获取注解类
+        Map<String, Object> beanMap;
+        beanMap = event.getApplicationContext().getBeansWithAnnotation(DtsListener.class);
+        //未扫描到不处理
+        if(beanMap.isEmpty()) {
+            return;
+        }
+        Map<String, RecordListener> finalBeanMap =  new HashMap<>();
+        beanMap.forEach((name, score) -> finalBeanMap.put(name, (RecordListener)score));
+        //初始化消费者
+        initDTSClient(finalBeanMap);
+    }
+
+    /**
+     * 初始化 初始化消费者
+     *
+     */
+    public void initDTSClient(Map<String, RecordListener> beans) {
+
+        ConsumerContext consumerContext = new ConsumerContext(
+                dtsProperties.getBrokerUrl(),
+                dtsProperties.getKafkaTopic(),
+                dtsProperties.getSid(),
+                dtsProperties.getUsername(),
+                dtsProperties.getPassword(),
+                dtsProperties.getInitCheckpoint(),
+                ConsumerContext.ConsumerSubscribeMode.ASSIGN);
+
+        //if this parameter is set, force to use the initCheckpoint to initial
+        consumerContext.setForceUseCheckpoint(dtsProperties.getIsForceUseInitCheckpoint());
+
+        //add user store
+        consumerContext.setUserRegisteredStore(new UserMetaStore(kxsCheckpointService));
+
+        dtsConsumer = new DefaultDTSConsumer(consumerContext);
+
+        dtsConsumer.addRecordListeners(beans);
+        dtsConsumer.start();
+    }
+
+
+    @Override
+    public void destroy() {
+        if(dtsConsumer != null) {
+            dtsConsumer.close();
+        }
+    }
+}

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

@@ -0,0 +1,64 @@
+package com.kxs.transfer.api.config;
+
+import lombok.Data;
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * dts数据订阅属性
+ *
+ * @author ： liukx
+ * @time ： 2020/9/22 - 17:03
+ */
+@Data
+@ConfigurationProperties(prefix = "spring.dts")
+public class DTSProperties {
+    /**
+     * 用户名
+     */
+    private String username;
+    /**
+     * 密码
+     */
+    private String password;
+    /**
+     * 消费组id [dts->数据消费列表->消费组ID]
+     */
+    private String sid;
+    /**
+     * 消费组名称[dts->数据消费列表->消费组名称]
+     */
+    private String groupName;
+    /**
+     * kafka topic配置 [dts->订阅配置->基本信息->订阅Topic]
+     */
+    private String kafkaTopic;
+    /**
+     * kafka 连接地址 [dts->订阅配置->网络->公私网]
+     */
+    private String brokerUrl;
+    /**
+     * 启动位点 : 这里表示消费的起点,如果是第一次启动则有效.
+     */
+    private String initCheckpoint = "1705998423";
+    /**
+     * 如果在启动时强制使用配置检查点。在检查点重置
+     */
+    private Boolean isForceUseInitCheckpoint = false;
+
+    /**
+     * 需要处理的表名
+     */
+    private List<String> filterTables;
+
+    /**
+     * 需要处理的库名
+     */
+    private List<String> filterDatabase;
+
+}

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

@@ -1,41 +1,44 @@
-//package com.kxs.transfer.api.config;
-//
-//import com.aliyun.dts.subscribe.clients.metastore.AbstractUserMetaStore;
-//import com.baomidou.mybatisplus.core.toolkit.Wrappers;
-//import com.kxs.transfer.api.model.KxsCheckpoint;
-//import com.kxs.transfer.api.service.KxsCheckpointService;
-//
-///**
-// * store the checkpoint data in the shared storage, such us database, shared file storage...
-// * this meta store need to be completed by consumer
-// */
-//public class UserMetaStore extends AbstractUserMetaStore {
-//
-//    private KxsCheckpointService kxsCheckpointService;
-//
-//    public UserMetaStore(KxsCheckpointService kxsCheckpointService) {
-//
-//        this.kxsCheckpointService = kxsCheckpointService;
-//    }
-//
-//    @Override
-//    protected void saveData(String groupID, String toStoreJson) {
-//
-//        System.out.println("消费节点" + groupID+ "保存checkpoint数据：" + toStoreJson);
-//        KxsCheckpoint kxsCheckpoint = new KxsCheckpoint();
-//        kxsCheckpoint.setGroupId(groupID);
-//        kxsCheckpoint.setCheckpoint(toStoreJson);
-//        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));
-//
-//
-//        return checkpoint.getCheckpoint();
-//    }
-//}
+package com.kxs.transfer.api.config;
+
+import com.aliyun.dts.subscribe.clients.metastore.AbstractUserMetaStore;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.kxs.transfer.api.model.KxsCheckpoint;
+import com.kxs.transfer.api.service.KxsCheckpointService;
+
+/**
+ * store the checkpoint data in the shared storage, such us database, shared file storage...
+ * this meta store need to be completed by consumer
+ * @author mac
+ */
+public class UserMetaStore extends AbstractUserMetaStore {
+
+    private final KxsCheckpointService kxsCheckpointService;
+
+    public UserMetaStore(KxsCheckpointService kxsCheckpointService) {
+
+        this.kxsCheckpointService = kxsCheckpointService;
+    }
+
+    @Override
+    protected void saveData(String groupID, String toStoreJson) {
+
+        System.out.println("消费节点" + groupID+ "保存checkpoint数据：" + toStoreJson);
+        KxsCheckpoint kxsCheckpoint = new KxsCheckpoint();
+        kxsCheckpoint.setGroupId(groupID);
+        kxsCheckpoint.setCheckpoint(toStoreJson);
+        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)
+                .last("limit 1"));
+
+
+        return checkpoint != null ? checkpoint.getCheckpoint() : null;
+    }
+}

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

@@ -0,0 +1,42 @@
+package com.kxs.transfer.api.controller;
+
+import com.alibaba.fastjson.JSON;
+import com.aliyun.dts.subscribe.clients.record.DefaultUserRecord;
+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 lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * <p>
+ *     DTS控制器
+ *     在此控制器里做数据分发
+ *     比如将数据分发到不同的数据源中
+ * </p>
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-25
+ */
+@RestController
+@Slf4j
+@RequiredArgsConstructor
+public class DtsController {
+
+    private final KxsDtsLogService kxsDtsLogService;
+
+    @DtsMsgListener
+    public void dtsListener(Long dataId, DMLData dmlData, DefaultUserRecord record){
+
+        //存储历史数据
+        KxsDtsLog kxsDtsLog = new KxsDtsLog();
+        kxsDtsLog.setDataId(dataId);
+        kxsDtsLog.setContent(JSON.toJSONString(dmlData));
+        kxsDtsLogService.save(kxsDtsLog);
+        log.info("消费dts的数据表信息:{}", dmlData);
+        record.commit(String.valueOf(record.getSourceTimestamp()));
+    }
+
+}

kxs-transfer/src/main/java/com/kxs/transfer/api/converter/MysqlFieldConverter.java

@@ -0,0 +1,388 @@
+package com.kxs.transfer.api.converter;
+
+import com.aliyun.dts.subscribe.clients.formats.avro.*;
+import com.aliyun.dts.subscribe.clients.formats.avro.Character;
+import com.aliyun.dts.subscribe.clients.formats.avro.Float;
+import com.aliyun.dts.subscribe.clients.formats.avro.Integer;
+import com.aliyun.dts.subscribe.clients.recordprocessor.FieldValue;
+
+import java.nio.ByteBuffer;
+
+import static java.nio.charset.StandardCharsets.US_ASCII;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+public class MysqlFieldConverter {
+
+    public static FieldValue convert(Field field, Object o) {
+        return DATA_ADAPTER[field.getDataTypeNumber()].getFieldValue(o);
+    }
+
+    static DataAdapter[] DATA_ADAPTER = new DataAdapter[256];
+
+    static {
+        DATA_ADAPTER[0] = new DecimalStringAdapter(); //Type.DECIMAL
+        DATA_ADAPTER[1] = new NumberStringAdapter(); //Type.INT8;
+        DATA_ADAPTER[2] = new NumberStringAdapter(); //Type.INT16;
+        DATA_ADAPTER[3] = new NumberStringAdapter(); //Type.INT32;
+
+        DATA_ADAPTER[4] = new DoubleStringAdapter(); //Type.FLOAT
+        DATA_ADAPTER[5] = new DoubleStringAdapter(); //Type.DOUBLE
+
+        DATA_ADAPTER[6] = new UTF8StringEncodeAdapter(); //Type.NULL
+
+        DATA_ADAPTER[7] = new TimestampStringAdapter(); //Type.TIMESTAMP
+        DATA_ADAPTER[8] = new NumberStringAdapter(); //Type.INT64
+        DATA_ADAPTER[9] = new NumberStringAdapter(); //Type.INT24
+
+        DATA_ADAPTER[10] = new DateAdapter(); //Type.DATE
+        DATA_ADAPTER[11] = new TimeAdapter(); //Type.TIME
+        DATA_ADAPTER[12] = new DateTimeAdapter(); //Type.DATETIME
+        DATA_ADAPTER[13] = new YearAdapter(); //Type.YEAR
+        DATA_ADAPTER[14] = new DateTimeAdapter(); //Type.DATETIME
+        DATA_ADAPTER[15] = new CharacterAdapter(); 	//Type.STRING
+        DATA_ADAPTER[16] = new NumberStringAdapter(); //Type.BIT
+
+        DATA_ADAPTER[255] = new GeometryAdapter(); 	//Type.GEOMETRY;
+        DATA_ADAPTER[254] = new CharacterAdapter(); //Type.STRING;
+        DATA_ADAPTER[253] = new CharacterAdapter(); //Type.STRING;
+
+        DATA_ADAPTER[252] = new BinaryAdapter(); //Type.BLOB;
+        DATA_ADAPTER[251] = new BinaryAdapter(); //Type.BLOB;
+        DATA_ADAPTER[250] = new BinaryAdapter(); //Type.BLOB;
+        DATA_ADAPTER[249] = new BinaryAdapter(); //Type.BLOB;
+
+        DATA_ADAPTER[246] = new DecimalStringAdapter(); //Type.DECIMAL;
+
+        DATA_ADAPTER[248] = new TextObjectAdapter(); //Type.SET;
+        DATA_ADAPTER[247] = new TextObjectAdapter(); //Type.ENUM;
+        DATA_ADAPTER[245] = new TextObjectAdapter();  //Type.JSON;
+    }
+
+    static interface DataAdapter {
+
+        FieldValue getFieldValue(Object data);
+    }
+
+
+    static class UTF8StringEncodeAdapter implements DataAdapter {
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                byte[] bytes = ((String) data).getBytes(UTF_8);
+                fieldValue.setValue(bytes);
+            }
+            fieldValue.setEncoding("UTF8");
+            return fieldValue;
+        }
+    }
+
+    static class NumberStringAdapter implements DataAdapter {
+        @Override
+        public FieldValue getFieldValue(Object data) {
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                Integer integer = (Integer) data;
+                fieldValue.setValue(integer.getValue().getBytes(US_ASCII));
+            }
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+    }
+
+    static class DecimalStringAdapter implements DataAdapter {
+        @Override
+        public FieldValue getFieldValue(Object data) {
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                Decimal decimal = (Decimal) data;
+                fieldValue.setValue(decimal.getValue().getBytes(US_ASCII));
+            }
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+    }
+
+    static class DoubleStringAdapter implements DataAdapter {
+        @Override
+        public FieldValue getFieldValue(Object data) {
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                Float aFloat = (Float) data;
+                fieldValue.setValue(Double.toString(aFloat.getValue()).getBytes(US_ASCII));
+            }
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+    }
+
+    static class TimestampStringAdapter implements DataAdapter {
+
+        static String[] MILLIS_PREFIX = new String[]{"", "0", "00", "000", "0000", "00000", "000000"};
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                StringBuilder timestampBuilder = new StringBuilder(64);
+
+                Timestamp timestamp = (Timestamp) data;
+
+                timestampBuilder.append(timestamp.getTimestamp());
+                if (null != timestamp.getMillis() && timestamp.getMillis() > 0) {
+                    timestampBuilder.append('.');
+                    String millis = java.lang.Integer.toString(timestamp.getMillis());
+                    timestampBuilder.append(MILLIS_PREFIX[6 - millis.length()]).append(millis);
+                }
+
+                fieldValue.setValue(timestampBuilder.toString().getBytes(US_ASCII));
+            }
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+    }
+
+    static abstract class AbstractDateTimeAdapter implements DataAdapter {
+
+        void encodeDate(DateTime dateTime, byte[] out, int position) {
+            if (null != dateTime && null != out) {
+                out[position] = (byte) ('0' + (dateTime.getYear() / 1000));
+                out[position + 1] = (byte) ('0' + (dateTime.getYear() % 1000 / 100));
+                out[position + 2] = (byte) ('0' + (dateTime.getYear() % 100 / 10));
+                out[position + 3] = (byte) ('0' + (dateTime.getYear() % 10));
+                out[position + 4] = '-';
+                out[position + 5] = (byte) ('0' + (dateTime.getMonth() / 10));
+                out[position + 6] = (byte) ('0' + (dateTime.getMonth() % 10));
+                out[position + 7] = '-';
+                out[position + 8] = (byte) ('0' + (dateTime.getDay() / 10));
+                out[position + 9] = (byte) ('0' + (dateTime.getDay() % 10));
+            }
+        }
+
+        void encodeTime(DateTime dateTime, byte[] out, int position) {
+            if (null != dateTime && null != out) {
+                out[position + 0] = (byte) ('0' + (dateTime.getHour() / 10));
+                out[position + 1] = (byte) ('0' + (dateTime.getHour() % 10));
+                out[position + 2] = ':';
+                out[position + 3] = (byte) ('0' + (dateTime.getMinute() / 10));
+                out[position + 4] = (byte) ('0' + (dateTime.getMinute() % 10));
+                out[position + 5] = ':';
+                out[position + 6] = (byte) ('0' + (dateTime.getSecond() / 10));
+                out[position + 7] = (byte) ('0' + (dateTime.getSecond() % 10));
+            }
+        }
+
+        void encodeTimeMillis(DateTime dateTime, byte[] out, int position) {
+            if (null != dateTime.getMillis() && 0 != dateTime.getMillis()) {
+                int mills = dateTime.getMillis();
+                out[position] = '.';
+                out[position + 1] = (byte) ('0' + (mills / 100000));
+                mills %= 100000;
+                out[position + 2] = (byte) ('0' + (mills / 10000));
+                mills %= 10000;
+                out[position + 3] = (byte) ('0' + (mills / 1000));
+                mills %= 1000;
+                out[position + 4] = (byte) ('0' + (mills / 100));
+                mills %= 100;
+                out[position + 5] = (byte) ('0' + (mills / 10));
+                out[position + 6] = (byte) ('0' + (mills % 10));
+            }
+        }
+    }
+
+    static class DateAdapter extends AbstractDateTimeAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                DateTime dateTime = (DateTime) data;
+
+                byte[] date = new byte[10];
+                encodeDate(dateTime, date, 0);
+
+                fieldValue.setValue(date);
+            }
+
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+    }
+
+
+    static class TimeAdapter extends AbstractDateTimeAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                DateTime dateTime = (DateTime)data;
+
+                // 判断是否是负数
+                int head = 0;
+                if (dateTime.getHour() <= -100) {
+                    head = 2;
+                } else if ((dateTime.getHour() >= 100)
+                        || (dateTime.getHour() < 0)
+                        || (dateTime.getMinute() < 0)
+                        || (dateTime.getSecond() < 0)
+                        || ((null != dateTime.getMillis()) && (dateTime.getMillis() < 0))) {
+                    head = 1;
+                }
+                byte[] time;
+                // 毫秒位0忽略
+                if (null == dateTime.getMillis() || 0 == dateTime.getMillis()) {
+                    time = new byte[8 + head];
+                } else {
+                    time = new byte[15 + head];
+                }
+
+                int index = 0;
+                if (head > 0 && dateTime.getHour() <= 0) {
+                    dateTime.setHour(-dateTime.getHour());
+                    dateTime.setMinute(-dateTime.getMinute());
+                    dateTime.setSecond(-dateTime.getSecond());
+                    if (null != dateTime.getMillis()) {
+                        dateTime.setMillis(-dateTime.getMillis());
+                    }
+
+                    time[index++] = '-';
+                }
+
+                if (dateTime.getHour() >= 100) {
+                    time[index++] = (byte) ('0' + (dateTime.getHour() / 100));
+                    dateTime.setHour(dateTime.getHour() % 100);
+                }
+
+                encodeTime(dateTime, time, index);
+                encodeTimeMillis(dateTime, time, index + 8);
+
+                fieldValue.setValue(time);
+            }
+
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+    }
+
+    static class DateTimeAdapter extends AbstractDateTimeAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                DateTime dateTime = (DateTime) data;
+
+                byte[] time = null;
+                //忽略毫秒位值是0
+                if (null == dateTime.getMillis() || 0 == dateTime.getMillis()) {
+                    time = new byte[19];
+                } else {
+                    time = new byte[26];
+                }
+                encodeDate(dateTime, time, 0);
+                time[10] = ' ';
+                encodeTime(dateTime, time, 11);
+                encodeTimeMillis(dateTime, time, 19);
+
+                fieldValue.setValue(time);
+            }
+
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+
+    }
+
+
+    static class YearAdapter implements DataAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                DateTime dateTime = (DateTime) data;
+                fieldValue.setValue(java.lang.Integer.toString(dateTime.getYear()).getBytes(US_ASCII));
+            }
+
+            fieldValue.setEncoding("ASCII");
+            return fieldValue;
+        }
+
+    }
+
+    static class CharacterAdapter implements DataAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                Character character = (Character) data;
+                fieldValue.setValue(getBytes(character.getValue()));
+                fieldValue.setEncoding(character.getCharset());
+            } else {
+                fieldValue.setEncoding("ASCII");
+            }
+            return fieldValue;
+        }
+
+    }
+
+    static class GeometryAdapter implements DataAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                BinaryGeometry geometry = (BinaryGeometry) data;
+                fieldValue.setValue(getBytes(geometry.getValue()));
+            }
+            return fieldValue;
+        }
+    }
+
+
+    static class BinaryAdapter implements DataAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                BinaryObject binaryObject = (BinaryObject) data;
+                fieldValue.setValue(getBytes(binaryObject.getValue()));
+            }
+            return fieldValue;
+        }
+    }
+
+
+    static class TextObjectAdapter implements DataAdapter {
+
+        @Override
+        public FieldValue getFieldValue(Object data) {
+
+            FieldValue fieldValue = new FieldValue();
+            if (null != data) {
+                TextObject textObject = (TextObject) data;
+                byte[] bytes = textObject.getValue().getBytes(UTF_8);
+                fieldValue.setValue(bytes);
+            }
+            fieldValue.setEncoding("UTF8");
+            return fieldValue;
+        }
+    }
+
+    static byte[] getBytes(ByteBuffer origin) {
+        byte[] ret = new byte[origin.remaining()];
+        origin.get(ret);
+        return ret;
+    }
+}

kxs-transfer/src/main/java/com/kxs/transfer/api/listener/MysqlConsumerListener.java

@@ -0,0 +1,294 @@
+package com.kxs.transfer.api.listener;
+
+import com.alibaba.fastjson.JSON;
+import com.aliyun.dts.subscribe.clients.common.FieldEntryHolder;
+import com.aliyun.dts.subscribe.clients.common.NullableOptional;
+import com.aliyun.dts.subscribe.clients.common.RecordListener;
+import com.aliyun.dts.subscribe.clients.formats.avro.Field;
+import com.aliyun.dts.subscribe.clients.formats.avro.Record;
+import com.aliyun.dts.subscribe.clients.record.DefaultUserRecord;
+import com.aliyun.dts.subscribe.clients.record.OperationType;
+import com.aliyun.dts.subscribe.clients.record.RecordSchema;
+import com.kxs.transfer.api.annotation.DtsListener;
+import com.kxs.transfer.api.annotation.DtsMsgListener;
+import com.kxs.transfer.api.config.DTSProperties;
+import com.kxs.transfer.api.converter.MysqlFieldConverter;
+import com.kxs.transfer.api.model.table.DMLData;
+import com.kxs.transfer.api.model.table.FieldData;
+import com.kxs.transfer.api.util.SpringContextUtil;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.util.*;
+
+
+/**
+ * 消费dts的数据监听器
+ *
+ * @author ： liukx
+ * @time ： 2020/9/22 - 17:22
+ */
+@Component
+@RequiredArgsConstructor
+@Slf4j
+@DtsListener
+@EnableConfigurationProperties(DTSProperties.class)
+public class MysqlConsumerListener implements RecordListener {
+
+    private final DTSProperties dtsProperties;
+
+
+    /**
+     * 对应的java属性类型
+     */
+    private static final Class[] FIELD_CLASS = new Class[256];
+
+    static {
+
+        FIELD_CLASS[0] = BigDecimal.class;
+
+        FIELD_CLASS[1] = Integer.class;
+        //Type.INT16;
+        FIELD_CLASS[2] = Integer.class;
+        //Type.INT32;
+        FIELD_CLASS[3] = Integer.class;
+
+        //Type.FLOAT
+        FIELD_CLASS[4] = Double.class;
+        //Type.DOUBLE
+        FIELD_CLASS[5] = Double.class;
+
+        //Type.NULL
+        FIELD_CLASS[6] = String.class;
+
+        //Type.TIMESTAMP
+        FIELD_CLASS[7] = Date.class;
+
+        //Type.INT64
+        FIELD_CLASS[8] = Integer.class;
+        //Type.INT24
+        FIELD_CLASS[9] = Integer.class;
+
+        //Type.DATE
+        FIELD_CLASS[10] = Date.class;
+        //Type.TIME
+        FIELD_CLASS[11] = Date.class;
+        //Type.DATETIME
+        FIELD_CLASS[12] = Date.class;
+        //Type.YEAR
+        FIELD_CLASS[13] = Date.class;
+        //Type.DATETIME
+        FIELD_CLASS[14] = Date.class;
+        //Type.STRING
+        FIELD_CLASS[15] = String.class;
+        //Type.BIT
+        FIELD_CLASS[16] = Integer.class;
+
+        FIELD_CLASS[255] = String.class;    //Type.GEOMETRY;
+        FIELD_CLASS[254] = String.class; //Type.STRING;
+        FIELD_CLASS[253] = String.class; //Type.STRING;
+
+        FIELD_CLASS[252] = String.class; //Type.BLOB;
+        FIELD_CLASS[251] = String.class; //Type.BLOB;
+        FIELD_CLASS[250] = String.class; //Type.BLOB;
+        FIELD_CLASS[249] = String.class; //Type.BLOB;
+
+        FIELD_CLASS[246] = Long.class; //Type.DECIMAL;
+
+        FIELD_CLASS[248] = String.class; //Type.SET;
+        FIELD_CLASS[247] = String.class; //Type.ENUM;
+        FIELD_CLASS[245] = String.class;  //Type.JSON;
+    }
+
+
+    @Override
+    public void consume(DefaultUserRecord record) {
+
+        OperationType operationType = record.getOperationType();
+        if(operationType.equals(OperationType.INSERT)
+                || operationType.equals(OperationType.UPDATE)
+                || operationType.equals(OperationType.DELETE)
+                || operationType.equals(OperationType.DDL)
+                || operationType.equals(OperationType.HEARTBEAT)) {
+
+            Record avroRecord = record.getAvroRecord();
+            RecordSchema schema = record.getSchema();
+            //库名
+            NullableOptional<String> databaseName = schema.getDatabaseName();
+            NullableOptional<String> tableName = schema.getTableName();
+            //过滤空库数据 过滤空表数据
+            if (!databaseName.isPresent() || !tableName.isPresent()) {
+                record.commit(String.valueOf(record.getSourceTimestamp()));
+                return;
+            }
+            //过滤库和表
+            if (!dtsProperties.getFilterDatabase().contains(databaseName.get())
+                    || !dtsProperties.getFilterTables().contains(tableName.get())) {
+                record.commit(String.valueOf(record.getSourceTimestamp()));
+                return;
+            }
+            //此参数是否唯一过滤数据，还有待考证
+            Long dataId = avroRecord.getId();
+            DMLData dmlData = parseDML(record);
+
+            SpringContextUtil.invokeMethodByAnnotation(DtsMsgListener.class,dataId, dmlData, record);
+
+        }
+
+    }
+
+    private DMLData parseDML(DefaultUserRecord record) {
+
+        RecordSchema schema = record.getSchema();
+        //库名
+        NullableOptional<String> databaseName = schema.getDatabaseName();
+        //表名
+        NullableOptional<String> tableName = schema.getTableName();
+
+        OperationType operationType = record.getOperationType();
+
+        //组装实体数据
+        Record avroRecord = record.getAvroRecord();
+        DMLData dmlData = new DMLData();
+        dmlData.setOperation(operationType);
+        dmlData.setDatabaseName(databaseName.get());
+        dmlData.setTableName(tableName.get());
+        dmlData.setSourceTimestamp(record.getSourceTimestamp());
+
+        List<String> changeFieldList = new ArrayList<>();
+        Map<String, FieldData> fieldDataMap = parseUpdateField(avroRecord, changeFieldList);
+        if (operationType == OperationType.INSERT || operationType == OperationType.UPDATE) {
+            Map<String, Object> validFieldDataMap = parseValidUpdateField(fieldDataMap, changeFieldList);
+            dmlData.setValidFieldDataMap(validFieldDataMap);
+        }
+        dmlData.setId(getFieldValue(fieldDataMap, "Id"));
+        dmlData.setFieldDataMap(fieldDataMap);
+        dmlData.setChangeFieldList(changeFieldList);
+        dmlData.setSourceTimestamp(record.getSourceTimestamp());
+        return dmlData;
+    }
+
+    /**
+     * 解析出有效数据
+     *
+     * @param fieldDataMap    现场数据映射
+     * @param changeFieldList 更改字段列表
+     * @return {@link Map}<{@link String}, {@link Object}>
+     */
+    public static Map<String, Object> parseValidUpdateField(Map<String, FieldData> fieldDataMap, List<String> changeFieldList) {
+        Map<String, Object> res = new HashMap<>(16);
+        for (String f : changeFieldList) {
+            FieldData fieldData = fieldDataMap.get(f);
+            res.put(f, fieldData.getValue());
+        }
+        if(fieldDataMap.get("Id") != null){
+            res.put("id", fieldDataMap.get("Id").getValue());
+        }
+        res.put("id", fieldDataMap.get("Id").getValue());
+        return res;
+    }
+
+    /**
+     * 获取数据的值
+     *
+     * @param fieldDataMap 数据对象
+     * @param fieldName    数据字段名称
+     * @return {@link String}
+     */
+    private static String getFieldValue(Map<String, FieldData> fieldDataMap, String fieldName) {
+        FieldData fieldData = fieldDataMap.get(fieldName);
+        if (fieldData != null) {
+            Object value = fieldData.getValue();
+            if (value != null) {
+                return value.toString();
+            }
+
+            Object oldValue = fieldData.getOldValue();
+            if (oldValue != null) {
+                return oldValue.toString();
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 解析更新字段
+     * 解析修改类型
+     *
+     * @param record          记录
+     * @param changeFieldList 更改字段列表
+     * @return {@link Map}<{@link String}, {@link FieldData}>
+     */
+    public static Map<String, FieldData> parseUpdateField(Record record, List<String> changeFieldList) {
+        List<Field> fields = (List<Field>) record.getFields();
+        FieldEntryHolder[] fieldArray = getFieldEntryHolder(record);
+        FieldEntryHolder before = fieldArray[0];
+        FieldEntryHolder after = fieldArray[1];
+
+        Map<String, FieldData> fieldDataMap = new LinkedHashMap<>();
+        if (null != fields) {
+            Iterator<Field> fieldIterator = fields.iterator();
+            while (fieldIterator.hasNext() && before.hasNext() && after.hasNext()) {
+                FieldData fieldData = new FieldData();
+                Field field = fieldIterator.next();
+                fieldData.setField(field.getName());
+
+                Class fieldType = FIELD_CLASS[field.getDataTypeNumber()];
+                fieldData.setDataType(fieldType);
+
+                Object toPrintBefore = before.take();
+                if (toPrintBefore != null) {
+                    Object beforeValue = getTypeValue(field, toPrintBefore);
+                    fieldData.setOldValue(beforeValue);
+                }
+
+                Object toPrintAfter = after.take();
+                if (toPrintAfter != null) {
+                    Object afterValue = getTypeValue(field, toPrintAfter);
+                    fieldData.setValue(afterValue);
+                }
+
+                if (fieldData.getValue() != null) {
+                    if (fieldData.getOldValue() != null) {
+                        if (!fieldData.getValue().equals(fieldData.getOldValue())) {
+                            changeFieldList.add(field.getName());
+                        }
+                    } else {
+                        changeFieldList.add(field.getName());
+                    }
+                } else if (fieldData.getOldValue() != null) {
+                    changeFieldList.add(field.getName());
+                }
+
+
+                fieldDataMap.put(field.getName(), fieldData);
+            }
+        }
+        return fieldDataMap;
+    }
+    private static Object getTypeValue(Field field, Object toPrintBefore) {
+
+        Object text = MysqlFieldConverter.convert(field, toPrintBefore).toString();
+
+        if (text != null && field.getDataTypeNumber() == 7) {
+            text = LocalDateTime.parse(text.toString());
+        } else if (text != null && field.getDataTypeNumber() == 245) {
+            text = JSON.parseObject(text.toString(), Map.class);
+        }
+        return text;
+    }
+
+    private static FieldEntryHolder[] getFieldEntryHolder(Record record) {
+        // this is a simple impl, may exist unhandled situation
+        FieldEntryHolder[] fieldArray = new FieldEntryHolder[2];
+
+        fieldArray[0] = new FieldEntryHolder((List<Object>) record.getBeforeImages());
+        fieldArray[1] = new FieldEntryHolder((List<Object>) record.getAfterImages());
+
+        return fieldArray;
+    }
+}

kxs-transfer/src/main/java/com/kxs/transfer/api/mapper/KxsDtsLogMapper.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.KxsDtsLog;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据同步记录日志表(KxsDtsLog)表数据库访问层
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+@Mapper
+@DS("sys")
+public interface KxsDtsLogMapper extends BaseMapper<KxsDtsLog> {
+
+}
+

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

@@ -0,0 +1,73 @@
+package com.kxs.transfer.api.model;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+
+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;
+
+/**
+ * 数据同步记录日志表(KxsDtsLog)表实体类
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+public class KxsDtsLog extends Model<KxsDtsLog> implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    
+    
+    /**
+     * 主键
+     */
+    @Schema(description = "主键")
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+
+    /**
+     * 数据ID
+     */
+    @Schema(description = "数据ID")
+    private Long dataId;
+
+
+    /**
+     * 原始数据
+     */
+    @Schema(description = "原始数据")
+    private String content;
+
+
+    /**
+     * 新增时间
+     */
+    @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/model/table/DMLData.java

@@ -0,0 +1,41 @@
+package com.kxs.transfer.api.model.table;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * 增删改操作实体
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-23
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+public class DMLData extends TableData {
+
+    /**
+     * 字段数据信息
+     */
+    private Map<String, FieldData> fieldDataMap;
+
+    /**
+     * 变更字段
+     */
+    private List<String> changeFieldList;
+
+    /**
+     * 数据的主键
+     */
+    private String id;
+    
+    /**
+     * 有效信息：增是全部字段，改是改动字段加主键，删是不处理
+     */
+    private Map<String, Object> validFieldDataMap;
+
+
+}

kxs-transfer/src/main/java/com/kxs/transfer/api/model/table/FieldData.java

@@ -0,0 +1,35 @@
+package com.kxs.transfer.api.model.table;
+
+
+import lombok.Data;
+
+/**
+ * 字段属性
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-23
+ */
+@Data
+public class FieldData {
+
+    /**
+     * 字段名称
+     */
+    private String field;
+
+    /**
+     * 当前字段值
+     */
+    private Object value;
+
+    /**
+     * 字段类型
+     */
+    private Class dataType;
+
+    /**
+     * 老的值
+     */
+    private Object oldValue;
+
+}

kxs-transfer/src/main/java/com/kxs/transfer/api/model/table/TableData.java

@@ -0,0 +1,38 @@
+package com.kxs.transfer.api.model.table;
+
+
+import com.aliyun.dts.subscribe.clients.formats.avro.Operation;
+import com.aliyun.dts.subscribe.clients.record.OperationType;
+import lombok.Data;
+
+
+/**
+ * 表结构
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-23
+ */
+@Data
+public class TableData {
+
+    /**
+     * 数据库名称
+     */
+    private String databaseName;
+    /**
+     * 表名称
+     */
+    private String tableName;
+
+    /**
+     * 操作类型
+     */
+    private OperationType operation;
+
+    /**
+     * 来源的时间戳
+     */
+    private Long sourceTimestamp;
+
+
+}

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

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

kxs-transfer/src/main/java/com/kxs/transfer/api/service/impl/KxsDtsLogServiceImpl.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.KxsDtsLogMapper;
+import com.kxs.transfer.api.model.KxsDtsLog;
+import com.kxs.transfer.api.service.KxsDtsLogService;
+import org.springframework.stereotype.Service;
+
+/**
+ * 数据同步记录日志表(KxsDtsLog)表服务实现类
+ *
+ * @author 系统
+ * @since 2024-01-25 14:25:31
+ */
+@Service("kxsDtsLogService")
+public class KxsDtsLogServiceImpl extends ServiceImpl<KxsDtsLogMapper, KxsDtsLog> implements KxsDtsLogService {
+
+}
+

kxs-transfer/src/main/java/com/kxs/transfer/api/util/SpringContextUtil.java

@@ -0,0 +1,115 @@
+package com.kxs.transfer.api.util;
+import lombok.Getter;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Spring 上下文实用程序
+ *
+ * @author 没秃顶的码农
+ * @date 2024-01-25
+ */
+@Component
+public class SpringContextUtil implements ApplicationContextAware {
+
+    //获取applicationContext
+    @Getter
+    private static ApplicationContext applicationContext;
+ 
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        SpringContextUtil.applicationContext = applicationContext;
+    }
+
+    //通过name获取 Bean.
+    public static Object getBean(String name) {
+        return getApplicationContext().getBean(name);
+    }
+ 
+    //通过class获取Bean.
+    public static <T> T getBean(Class<T> clazz) {
+        return getApplicationContext().getBean(clazz);
+    }
+ 
+    //通过name,以及Clazz返回指定的Bean
+    public static <T> T getBean(String name, Class<T> clazz) {
+        return getApplicationContext().getBean(name, clazz);
+    }
+
+    /**
+     * 通过注解调用方法
+     * 通过方法注解代理执行方法
+     *
+     * @param annotation 注解
+     * @param args       参数
+     */
+    public static void invokeMethodByAnnotation(Class<? extends Annotation> annotation,
+                                                   Object... args) {
+        List<Object> beans = getBeansByMethodAnnotation(annotation);
+ 
+        for (Object bean : beans) {
+            List<Method> methods = getMethodsByAnnotation(bean, annotation);
+            for (Method method : methods) {
+                try {
+                    method.invoke(bean,args);
+                } catch (IllegalAccessException | InvocationTargetException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    /**
+     * 通过方法注解获取 Bean
+     *
+     * @param annotation 注解
+     * @return {@link List}<{@link Object}>
+     */
+    public static List<Object> getBeansByMethodAnnotation(Class<? extends Annotation> annotation) {
+        List<Object> list = new ArrayList<>();
+        String[] beanNames = getApplicationContext().getBeanDefinitionNames();
+ 
+        for (String beanName : beanNames) {
+            Object bean = getBean(beanName);
+            List<Method> methodsByAnnotation = getMethodsByAnnotation(bean, annotation);
+ 
+            if (methodsByAnnotation.isEmpty()){
+                //方法未匹配则跳过
+                continue;
+            }
+ 
+            list.add(bean);
+        }
+ 
+        return list;
+    }
+
+    /**
+     * 通过注解获取方法
+     *
+     * @param bean       豆
+     * @param annotation 注解
+     * @return {@link List}<{@link Method}>
+     */
+    private static  List<Method> getMethodsByAnnotation(Object bean, Class<? extends Annotation> annotation) {
+        List<Method> methodList = new ArrayList<>();
+        Method[] methods = bean.getClass().getMethods();
+
+        for (Method item : methods) {
+            Annotation annotation1 = item.getAnnotation(annotation);
+            if (annotation1!=null) {
+                methodList.add(item);
+            }
+        }
+ 
+        return methodList;
+    }
+ 
+}