no message

.drone.yml

@@ -35,10 +35,12 @@ steps:
       - mvn clean package -Dmaven.test.skip=true -P test
 
       - ls -l /drone/src/kxs-product/target # 查看当前目录
-#      - cp /drone/src/kxs-gateway/target/*.jar /drone/src/dist/ && cp -r /drone/src/kxs-gateway/target/lib /drone/src/dist/ && cp -r /drone/src/kxs-gateway/target/config /drone/src/dist/ # 复制文件
-
+      # - cp /drone/src/kxs-gateway/target/*.jar /drone/src/dist/ && cp -r /drone/src/kxs-gateway/target/lib /drone/src/dist/ && cp -r /drone/src/kxs-gateway/target/config /drone/src/dist/ # 复制文件
+    when: success # 当前步骤成功时执行
   - name: scp-project
     image: appleboy/drone-scp
+    depends_on: # 依赖的步骤
+      - build-project
     settings:
       host: 172.27.251.48
       username: root
@@ -52,6 +54,7 @@ steps:
         - /drone/src/kxs-gateway/target/lib
         - /drone/src/kxs-gateway/target/config
         - echo ====结束拷贝=======
+
 trigger:
   branch:
     include:
@@ -60,34 +63,34 @@ trigger:
     include:
       - push
 
-# ---
+  # ---
+
+  # kind: pipeline
+  # type: docker
+  # name: deploy
+
+  # depends_on: # 依赖build管道
+  #   - build
+
+  # clone:
+  #   disable: true # 禁用拉取
 
-#kind: pipeline
-#type: docker
-#name: deploy
-#
-#depends_on: # 依赖build管道
-#  - build
-#
-#clone:
-#  disable: true # 禁用拉取
-#
-#steps:
-#  - name: scp-project
-#    image: appleboy/drone-scp
-#    settings:
-#      host: 172.27.251.48
-#      username: root
-#      password: esG1rZb2ZvalCbOc
-#      port: 22
-#      command_timeout: 2m
-#      target: /home/ABServer/webroot/kxs-gateway/
-#      source:
-#        - echo ====开始拷贝=======
-#        - /drone/src/kxs-gateway/target/*.jar
-#        - /drone/src/kxs-gateway/target/lib
-#        - /drone/src/kxs-gateway/target/config
-#        - echo ====结束拷贝=======
+  # steps:
+  #   - name: scp-project
+  #     image: appleboy/drone-scp
+  #     settings:
+  #       host: 172.27.251.48
+  #       username: root
+  #       password: esG1rZb2ZvalCbOc
+  #       port: 22
+  #       command_timeout: 2m
+  #       target: /home/ABServer/webroot/kxs-gateway/
+  #       source:
+  #         - echo ====开始拷贝=======
+  #         - /drone/src/kxs-gateway/target/*.jar
+  #         - /drone/src/kxs-gateway/target/lib
+  #         - /drone/src/kxs-gateway/target/config
+  #         - echo ====结束拷贝=======
 
   # - name: deploy-project
   #   image: appleboy/drone-ssh
@@ -112,10 +115,10 @@ trigger:
   #       - echo ====部署成功=======
 
 
-#trigger:
-#  branch:
-#    include:
-#      - master
-#  event:
-#    include:
-#      - push
+# trigger:
+#   branch:
+#     include:
+#       - master
+#   event:
+#     include:
+#       - push

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

@@ -1,91 +1,91 @@
-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.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);
+//    }
+//
+//}

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

@@ -0,0 +1,41 @@
+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();
+    }
+}