add spi

dbsyncer-connector/dbsyncer-connector-base/src/test/java/ConnectionTest.java

@@ -234,8 +234,8 @@ public class ConnectionTest {
         final DatabaseConnectorInstance connectorInstance = new DatabaseConnectorInstance(createMysqlConfig());
 
         long begin = Instant.now().toEpochMilli();
-        final int threadSize = 10;
-        final int num = 1000;
+        final int threadSize = 1000;
+        final int num = 100;
         final ExecutorService pool = Executors.newFixedThreadPool(threadSize);
         final CountDownLatch latch = new CountDownLatch(threadSize);
         final String insert = "INSERT INTO `vote_records_test` (`id`, `user_id`, `vote_num`, `group_id`, `status`, `create_time`) VALUES (?, ?, ?, ?, ?, ?)";

dbsyncer-parser/src/main/java/org/dbsyncer/parser/ParserSupportConfiguration.java

@@ -4,10 +4,15 @@
 package org.dbsyncer.parser;
 
 import org.dbsyncer.parser.flush.impl.TableGroupBufferActuator;
+import org.dbsyncer.sdk.spi.TableGroupBufferActuatorService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.ServiceLoader;
+
 /**
  * @Author AE86
  * @Version 1.0.0
@@ -16,9 +21,23 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 public class ParserSupportConfiguration {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
     @Bean
     @ConditionalOnMissingBean
     public TableGroupBufferActuator tableGroupBufferActuator() {
+        ServiceLoader<TableGroupBufferActuatorService> services = ServiceLoader.load(TableGroupBufferActuatorService.class, Thread.currentThread().getContextClassLoader());
+        for (TableGroupBufferActuatorService s : services) {
+            try {
+                TableGroupBufferActuatorService service = s.getClass().newInstance();
+                if (service instanceof TableGroupBufferActuator) {
+                    return (TableGroupBufferActuator) service;
+                }
+            } catch (Exception e) {
+                logger.error(e.getMessage(), e);
+                throw new ParserException("获取TableGroupBufferActuator异常.");
+            }
+        }
         return new TableGroupBufferActuator();
     }
 }

dbsyncer-web/src/main/resources/application.properties

@@ -33,17 +33,17 @@ dbsyncer.parser.general.buffer-period-millisecond=300
 # 每个驱动最多可分配的表执行器个数
 dbsyncer.parser.table.group.max-buffer-actuator-size=20
 # [TableGroupBufferActuator]线程数
-dbsyncer.parser.table.group.thread-core-size=2
+dbsyncer.parser.table.group.thread-core-size=1
 # [TableGroupBufferActuator]最大线程数
-dbsyncer.parser.table.group.max-thread-size=10
+dbsyncer.parser.table.group.max-thread-size=1
 # [TableGroupBufferActuator]线程池队列
 dbsyncer.parser.table.group.thread-queue-capacity=16
 # [TableGroupBufferActuator]单次执行任务数
 dbsyncer.parser.table.group.buffer-writer-count=1000
 # [TableGroupBufferActuator]每次消费缓存队列的任务数
-dbsyncer.parser.table.group.buffer-pull-count=20000
+dbsyncer.parser.table.group.buffer-pull-count=1000
 # [TableGroupBufferActuator]缓存队列容量
-dbsyncer.parser.table.group.buffer-queue-capacity=40000
+dbsyncer.parser.table.group.buffer-queue-capacity=10000
 # [TableGroupBufferActuator]定时消费缓存队列间隔(毫秒)
 dbsyncer.parser.table.group.buffer-period-millisecond=300