add storage mysql

dbsyncer-connector/src/main/java/org/dbsyncer/connector/util/JDBCUtil.java

@@ -12,7 +12,7 @@ public abstract class JDBCUtil {
 
     private final static Logger logger = LoggerFactory.getLogger(JDBCUtil.class);
 
-    public static Connection getConnection(String driver, String url, String username, String password) throws ClassNotFoundException, SQLException {
+    public static Connection getConnection(String driver, String url, String username, String password) throws SQLException {
         // com.mysql.jdbc.JDBC4Connection 
         // 不需要显式调用 Class.forName(driver), DriverManager.getConnection会自动加载合适的驱动
         return DriverManager.getConnection(url, username, password);
@@ -43,16 +43,4 @@ public abstract class JDBCUtil {
         close(conn);
     }
 
-    //    public static void main(String[] args) {
-    //        String url = "jdbc:mysql://10.238.206.222:13306/test?seUnicode=true&characterEncoding=UTF8&useSSL=true";
-    //        String username = "root";
-    //        String password = "123";
-    //        Connection connection = JDBCUtil.getConnection("com.mysql.jdbc.Driver", url, username, password);
-    //        if(connection!=null){
-    //            JDBCUtil.close(connection);
-    //        }else{
-    //            logger.error("can not connect url.");
-    //        }
-    //    }
-
-}
+}

dbsyncer-storage/pom.xml

@@ -26,6 +26,13 @@
             <version>${project.parent.version}</version>
         </dependency>
 
+        <!-- Connector 连接器 -->
+        <dependency>
+            <groupId>org.ghi</groupId>
+            <artifactId>dbsyncer-connector</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
         <!-- 全文检索 -->
         <dependency>
             <groupId>org.apache.lucene</groupId>

dbsyncer-storage/src/main/java/org/dbsyncer/storage/AbstractStorageService.java

@@ -6,6 +6,7 @@ import org.dbsyncer.storage.strategy.Strategy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.util.Assert;
@@ -19,7 +20,7 @@ import java.util.Map;
  * @version 1.0.0
  * @date 2019/11/16 1:28
  */
-public abstract class AbstractStorageService implements StorageService, ApplicationContextAware {
+public abstract class AbstractStorageService implements StorageService, ApplicationContextAware, DisposableBean {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 

dbsyncer-storage/src/main/java/org/dbsyncer/storage/support/DiskStorageServiceImpl.java

@@ -33,8 +33,8 @@ import java.util.stream.Collectors;
  * @version 1.0.0
  * @date 2019/9/10 23:22
  */
-@Component("diskStorageServiceImpl")
-@ConditionalOnProperty(value = "dbsyncer.storage.support.disk")
+@Component
+@ConditionalOnProperty(value = "dbsyncer.storage.support.disk", havingValue = "true")
 public class DiskStorageServiceImpl extends AbstractStorageService {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
@@ -160,4 +160,10 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
         return map.get(collectionId);
     }
 
+    @Override
+    public void destroy() throws Exception {
+        for (Map.Entry<String, Shard> m: map.entrySet()) {
+            m.getValue().close();
+        }
+    }
 }

dbsyncer-storage/src/main/java/org/dbsyncer/storage/support/MysqlStorageServiceImpl.java

@@ -1,13 +1,23 @@
 package org.dbsyncer.storage.support;
 
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.connector.util.DatabaseUtil;
 import org.dbsyncer.storage.AbstractStorageService;
+import org.dbsyncer.storage.enums.StorageEnum;
 import org.dbsyncer.storage.query.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 
-import java.io.IOException;
+import javax.annotation.PostConstruct;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * @author AE86
@@ -15,41 +25,101 @@ import java.util.Map;
  * @date 2019/9/10 23:22
  */
 @Component
-@ConditionalOnProperty(value = "dbsyncer.storage.support.mysql")
+@ConditionalOnProperty(value = "dbsyncer.storage.support.mysql", havingValue = "true")
+@ConfigurationProperties(prefix = "dbsyncer.storage.support.mysql")
 public class MysqlStorageServiceImpl extends AbstractStorageService {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    private JdbcTemplate jdbcTemplate;
+
+    private DatabaseConfig config;
+
+    private Set<String> tables = new HashSet();
+
+    private final Object createTableLock = new Object();
+
+    @PostConstruct
+    private void init() {
+        config = null == config ? new DatabaseConfig() : config;
+        config.setUrl(StringUtils.isNotBlank(config.getUrl()) ? config.getUrl()
+                : "jdbc:mysql://127.0.0.1:3306/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&useSSL=true");
+        config.setDriverClassName(
+                StringUtils.isNotBlank(config.getDriverClassName()) ? config.getDriverClassName() : "com.mysql.jdbc.Driver");
+        config.setUsername(StringUtils.isNotBlank(config.getUsername()) ? config.getUsername() : "root");
+        config.setPassword(StringUtils.isNotBlank(config.getPassword()) ? config.getPassword() : "123");
+        logger.info("url:{}", config.getUrl());
+        logger.info("driverClassName:{}", config.getDriverClassName());
+        logger.info("username:{}", config.getUsername());
+        logger.info("password:{}", config.getPassword());
+        jdbcTemplate = DatabaseUtil.getJdbcTemplate(config);
+
+        // 创建配置和日志表
+        createTableIfNotExist(StorageEnum.CONFIG.getType());
+        createTableIfNotExist(StorageEnum.LOG.getType());
+    }
+
     @Override
-    public List<Map> select(String collectionId, Query query) {
+    public List<Map> select(String table, Query query) {
+        createTableIfNotExist(table);
+
         return null;
     }
 
     @Override
-    public void insert(String collectionId, Map params) throws IOException {
+    public void insert(String table, Map params) {
+        createTableIfNotExist(table);
 
     }
 
     @Override
-    public void update(String collectionId, Map params) throws IOException {
+    public void update(String table, Map params) {
+        createTableIfNotExist(table);
 
     }
 
     @Override
-    public void delete(String collectionId, String id) throws IOException {
+    public void delete(String table, String id) {
+        createTableIfNotExist(table);
 
     }
 
     @Override
-    public void deleteAll(String collectionId) throws IOException {
+    public void deleteAll(String table) {
+        createTableIfNotExist(table);
 
     }
 
     @Override
-    public void insertLog(String collectionId, Map<String, Object> params) throws IOException {
+    public void insertLog(String table, Map<String, Object> params) {
+        createTableIfNotExist(table);
 
     }
 
     @Override
-    public void insertData(String collectionId, List<Map> list) throws IOException {
+    public void insertData(String table, List<Map> list) {
+        createTableIfNotExist(table);
+
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        DatabaseUtil.close(jdbcTemplate);
+    }
+
+    private void createTableIfNotExist(String table) {
+        synchronized (createTableLock) {
+            // 1、检查本地缓存
+            if (tables.contains(table)) {
+                return;
+            }
+
+            // 2、检查DB中是否创建
+            // show tables;
+        }
+    }
 
+    public void setConfig(DatabaseConfig config) {
+        this.config = config;
     }
 }

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

@@ -11,6 +11,10 @@ server.servlet.session.timeout=1800
 dbsyncer.storage.id=1
 dbsyncer.storage.support.disk=true
 #dbsyncer.storage.support.mysql=true
+#dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3306/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&useSSL=true
+#dbsyncer.storage.support.mysql.config.driverClassName=com.mysql.jdbc.Driver
+#dbsyncer.storage.support.mysql.config.username=root
+#dbsyncer.storage.support.mysql.config.password=123
 
 #monitor
 management.endpoints.web.base-path=/app