Merge pull request #1131 from rewerma/master

 rdb同步适配多数据源

client-adapter/README.md

@@ -19,14 +19,20 @@ client-adapter分为适配器和启动器两部分, 适配器为多个fat jar,
 
 启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为:
 ```
-canal-adapter-launcher.jar
+- bin
+    restart.sh
+    startup.bat
+    startup.sh
+    stop.sh
 - lib
     client-adapter.logger-1.1.1-jar-with-dependencies.jar
     client-adapter.hbase-1.1.1-jar-with-dependencies.jar
-- config
+    ...
+- conf
     application.yml
     - hbase
         mytest_person2.yml
+- logs
 ```
 以上目录结构最终会打包成 canal-adapter-launcher.tar.gz 压缩包
 
@@ -96,7 +102,7 @@ curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
 ```
 ### 2.3 启动canal-adapter示例
 #### 2.3.1 启动canal server (单机模式), 参考: [Canal QuickStart](https://github.com/alibaba/canal/wiki/QuickStart)
-#### 2.3.2 修改config/application.yml为:
+#### 2.3.2 修改conf/application.yml为:
 ```
 server:
   port: 8081
@@ -118,9 +124,9 @@ canal.conf:
     - outAdapters:
       - name: logger
 ```
-启动 canal-adapter-launcher.jar
+启动
 ```
-java -jar canal-adapter-launcher.jar
+bin/startup.sh
 ```
 
 ## 三、HBase适配器
@@ -161,7 +167,7 @@ adapter.conf:
 ```
 其中指定了一个HBase表映射文件: mytest_person.yml
 ### 3.2 适配器表映射文件
-修改 config/hbase/mytest_person.yml文件:
+修改 conf/hbase/mytest_person.yml文件:
 ```
 dataSourceKey: defaultDS            # 对应application.yml中的datasourceConfigs下的配置
 hbaseMapping:                       # mysql--HBase的单表映射配置

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java

@@ -1,7 +1,9 @@
 package com.alibaba.otter.canal.client.adapter.support;
 
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * 配置信息类
@@ -129,7 +131,9 @@ public class CanalClientConfig {
 
     public static class Group {
 
-        private List<OuterAdapterConfig> outAdapters; // 适配器列表
+        private List<OuterAdapterConfig>        outAdapters;                           // 适配器列表
+
+        private Map<String, OuterAdapterConfig> outAdaptersMap = new LinkedHashMap<>();
 
         public List<OuterAdapterConfig> getOutAdapters() {
             return outAdapters;
@@ -137,6 +141,17 @@ public class CanalClientConfig {
 
         public void setOutAdapters(List<OuterAdapterConfig> outAdapters) {
             this.outAdapters = outAdapters;
+            if (outAdapters != null) {
+                outAdapters.forEach(outAdapter -> outAdaptersMap.put(outAdapter.getKey(), outAdapter));
+            }
+        }
+
+        public Map<String, OuterAdapterConfig> getOutAdaptersMap() {
+            return outAdaptersMap;
+        }
+
+        public void setOutAdaptersMap(Map<String, OuterAdapterConfig> outAdaptersMap) {
+            this.outAdaptersMap = outAdaptersMap;
         }
     }
 

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java

@@ -40,6 +40,8 @@ public class ExtensionLoader<T> {
 
     private static final ConcurrentMap<Class<?>, Object>             EXTENSION_INSTANCES        = new ConcurrentHashMap<>();
 
+    private static final ConcurrentMap<String, Object>               EXTENSION_KEY_INSTANCES    = new ConcurrentHashMap<>();
+
     private final Class<?>                                           type;
 
     private final String                                             classLoaderPolicy;
@@ -116,6 +118,30 @@ public class ExtensionLoader<T> {
         return (T) instance;
     }
 
+    @SuppressWarnings("unchecked")
+    public T getExtension(String name, String key) {
+        if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null");
+        if ("true".equals(name)) {
+            return getDefaultExtension();
+        }
+        Holder<Object> holder = cachedInstances.get(name + "-" + key);
+        if (holder == null) {
+            cachedInstances.putIfAbsent(name + "-" + key, new Holder<>());
+            holder = cachedInstances.get(name + "-" + key);
+        }
+        Object instance = holder.get();
+        if (instance == null) {
+            synchronized (holder) {
+                instance = holder.get();
+                if (instance == null) {
+                    instance = createExtension(name, key);
+                    holder.set(instance);
+                }
+            }
+        }
+        return (T) instance;
+    }
+
     /**
      * 返回缺省的扩展，如果没有设置则返回<code>null</code>
      */
@@ -148,6 +174,45 @@ public class ExtensionLoader<T> {
         }
     }
 
+    @SuppressWarnings("unchecked")
+    private T createExtension(String name, String key) {
+        Class<?> clazz = getExtensionClasses().get(name);
+        if (clazz == null) {
+            throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+                                            + ")  could not be instantiated: class could not be found");
+        }
+        try {
+            T instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key);
+            if (instance == null) {
+                EXTENSION_KEY_INSTANCES.putIfAbsent(name + "-" + key, clazz.newInstance());
+                instance = (T) EXTENSION_KEY_INSTANCES.get(name + "-" + key);
+            }
+            return instance;
+        } catch (Throwable t) {
+            throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+                                            + ")  could not be instantiated: " + t.getMessage(),
+                t);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    // public T newInstance(String name) {
+    // Class<?> clazz = getExtensionClasses().get(name);
+    // if (clazz == null) {
+    // throw new IllegalStateException("Extension instance(name: " + name + ",
+    // class: " + type
+    // + ") could not be instantiated: class could not be found");
+    // }
+    // try {
+    // return (T) clazz.newInstance();
+    // } catch (Throwable t) {
+    // throw new IllegalStateException("Extension instance(name: " + name + ",
+    // class: " + type
+    // + ") could not be instantiated: " + t.getMessage(),
+    // t);
+    // }
+    // }
+
     private Map<String, Class<?>> getExtensionClasses() {
         Map<String, Class<?>> classes = cachedClasses.get();
         if (classes == null) {

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java

@@ -12,6 +12,8 @@ public class OuterAdapterConfig {
 
     private String              name;       // 适配器名称, 如: logger, hbase, es
 
+    private String              key;        // 适配器唯一键
+
     private String              hosts;      // 适配器内部的地址, 比如对应es该参数可以填写es的server地址
 
     private String              zkHosts;    // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址
@@ -26,6 +28,14 @@ public class OuterAdapterConfig {
         this.name = name;
     }
 
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
     public String getHosts() {
         return hosts;
     }

client-adapter/launcher/src/main/assembly/dev.xml

@@ -23,7 +23,7 @@
 		</fileSet>
 		<fileSet>
 			<directory>./src/main/resources</directory>
-			<outputDirectory>/config</outputDirectory>
+			<outputDirectory>/conf</outputDirectory>
 			<includes>
 				<include>**/*</include>
 
@@ -31,21 +31,21 @@
 		</fileSet>
 		<fileSet>
 			<directory>../elasticsearch/src/main/resources/es</directory>
-			<outputDirectory>/config/es</outputDirectory>
+			<outputDirectory>/conf/es</outputDirectory>
 			<includes>
 				<include>**/*</include>
 			</includes>
 		</fileSet>
         <fileSet>
             <directory>../hbase/src/main/resources/hbase</directory>
-            <outputDirectory>/config/hbase</outputDirectory>
+            <outputDirectory>/conf/hbase</outputDirectory>
             <includes>
                 <include>**/*</include>
             </includes>
         </fileSet>
 		<fileSet>
 			<directory>../rdb/src/main/resources/</directory>
-			<outputDirectory>/config</outputDirectory>
+			<outputDirectory>/conf</outputDirectory>
             <excludes>
                 <exclude>META-INF/**</exclude>
             </excludes>

client-adapter/launcher/src/main/bin/startup.sh

@@ -10,8 +10,6 @@ case "`uname`" in
 		;;
 esac
 base=${bin_abs_path}/..
-#canal_conf=$base/config/canal.properties
-#logback_configurationFile=$base/conf/logback.xml
 export LANG=en_US.UTF-8
 export BASE=$base
 

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java

@@ -123,7 +123,12 @@ public class CanalAdapterLoader {
 
     private void loadConnector(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
         try {
-            OuterAdapter adapter = loader.getExtension(config.getName());
+            OuterAdapter adapter;
+            if ("rdb".equalsIgnoreCase(config.getName())) {
+                adapter = loader.getExtension(config.getName(), config.getKey());
+            } else {
+                adapter = loader.getExtension(config.getName());
+            }
             ClassLoader cl = Thread.currentThread().getContextClassLoader();
             // 替换ClassLoader
             Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader());
@@ -133,6 +138,7 @@ public class CanalAdapterLoader {
             logger.info("Load canal adapter: {} succeed", config.getName());
         } catch (Exception e) {
             logger.error("Load canal adapter: {} failed", config.getName(), e);
+            System.exit(0);
         }
     }
 

client-adapter/launcher/src/main/resources/application.yml

@@ -11,16 +11,30 @@ spring:
     time-zone: GMT+8
     default-property-inclusion: non_null
 
-#canal.conf:
-#  canalServerHost: 127.0.0.1:11111
+canal.conf:
+  canalServerHost: 127.0.0.1:11111
 #  zookeeperHosts: slave1:2181
 #  bootstrapServers: slave1:6667 #or rocketmq
 #  flatMessage: true
-#  canalInstances:
-#  - instance: example
-#    groups:
-#    - outAdapters:
-#      - name: logger
+  canalInstances:
+  - instance: example
+    groups:
+    - outAdapters:
+      - name: logger
+#      - name: rdb
+#        key: oracle1
+#        properties:
+#          jdbc.driverClassName: oracle.jdbc.OracleDriver
+#          jdbc.url: jdbc:oracle:thin:@127.0.0.1:1521:orcl
+#          jdbc.username: mytest
+#          jdbc.password: m121212
+#      - name: rdb
+#        key: postgres1
+#        properties:
+#          jdbc.driverClassName: org.postgresql.Driver
+#          jdbc.url: jdbc:postgresql://localhost:5432/postgres
+#          jdbc.username: postgres
+#          jdbc.password: 121212
 #      - name: hbase
 #        properties:
 #          hbase.zookeeper.quorum: 127.0.0.1
@@ -37,18 +51,6 @@ spring:
 #    - groupId: g2
 #      outAdapters:
 #      - name: logger
-#      - name: oracle
-#        properties:
-#          jdbc.driverClassName: oracle.jdbc.OracleDriver
-#          jdbc.url: jdbc:oracle:thin:@127.0.0.1:49161:XE
-#          jdbc.username: mytest
-#          jdbc.password: m121212
-#      - name: postgresql
-#        properties:
-#          jdbc.driverClassName: org.postgresql.Driver
-#          jdbc.url: jdbc:postgresql://localhost:5432/postgres
-#          jdbc.username: postgres
-#          jdbc.password: 121212
 #  mqTopics:
 #  - mqMode: rocketmq
 #    topic: example
@@ -66,5 +68,4 @@ spring:
 #  adapterConfigs:
 #  - hbase/mytest_person2.yml
 #  - es/mytest_user.yml
-#  - oracle/mytest_user.yml
-#  - postgresql/mytest_user.yml
+#  - rdb/mytest_user.yml

client-adapter/rdb/pom.xml

@@ -27,30 +27,30 @@
             <scope>provided</scope>
         </dependency>
 
-        <dependency>
-            <groupId>mysql</groupId>
-            <artifactId>mysql-connector-java</artifactId>
-            <version>5.1.40</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.postgresql</groupId>
-            <artifactId>postgresql</artifactId>
-            <version>42.1.4</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.oracle</groupId>
-            <artifactId>ojdbc6</artifactId>
-            <version>11.2.0.4.0-atlassian-hosted</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.microsoft.sqlserver</groupId>
-            <artifactId>mssql-jdbc</artifactId>
-            <version>7.0.0.jre8</version>
-            <scope>test</scope>
-        </dependency>
+        <!--<dependency>-->
+            <!--<groupId>mysql</groupId>-->
+            <!--<artifactId>mysql-connector-java</artifactId>-->
+            <!--<version>5.1.40</version>-->
+            <!--<scope>test</scope>-->
+        <!--</dependency>-->
+        <!--<dependency>-->
+            <!--<groupId>org.postgresql</groupId>-->
+            <!--<artifactId>postgresql</artifactId>-->
+            <!--<version>42.1.4</version>-->
+            <!--<scope>test</scope>-->
+        <!--</dependency>-->
+        <!--<dependency>-->
+            <!--<groupId>com.oracle</groupId>-->
+            <!--<artifactId>ojdbc6</artifactId>-->
+            <!--<version>11.2.0.4.0-atlassian-hosted</version>-->
+            <!--<scope>test</scope>-->
+        <!--</dependency>-->
+        <!--<dependency>-->
+            <!--<groupId>com.microsoft.sqlserver</groupId>-->
+            <!--<artifactId>mssql-jdbc</artifactId>-->
+            <!--<version>7.0.0.jre8</version>-->
+            <!--<scope>test</scope>-->
+        <!--</dependency>-->
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java

@@ -19,12 +19,13 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService;
 import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService;
 import com.alibaba.otter.canal.client.adapter.support.*;
 
-public abstract class RdbAdapter implements OuterAdapter {
+@SPI("rdb")
+public class RdbAdapter implements OuterAdapter {
 
-    private static Logger                       logger             = LoggerFactory.getLogger(RdbAdapter.class);
+    private static Logger                       logger     = LoggerFactory.getLogger(RdbAdapter.class);
 
-    private volatile Map<String, MappingConfig> rdbMapping         = null;                                     // 文件名对应配置
-    private volatile Map<String, MappingConfig> mappingConfigCache = null;                                     // 库名-表名对应配置
+    private volatile Map<String, MappingConfig> rdbMapping = new HashMap<>();                          // 文件名对应配置
+    private Map<String, MappingConfig>          mappingConfigCache;                                    // 库名-表名对应配置
 
     private DruidDataSource                     dataSource;
 
@@ -32,20 +33,20 @@ public abstract class RdbAdapter implements OuterAdapter {
 
     @Override
     public void init(OuterAdapterConfig configuration) {
-        if (mappingConfigCache == null) {
-            synchronized (MappingConfig.class) {
-                if (mappingConfigCache == null) {
-                    SPI spi = this.getClass().getAnnotation(SPI.class);
-                    rdbMapping = MappingConfigLoader.load(spi.value());
-                    mappingConfigCache = new HashMap<>();
-                    for (MappingConfig mappingConfig : rdbMapping.values()) {
-                        mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
-                                               + mappingConfig.getDbMapping().getDatabase() + "."
-                                               + mappingConfig.getDbMapping().getTable(),
-                            mappingConfig);
-                    }
-                }
+        SPI spi = this.getClass().getAnnotation(SPI.class);
+        Map<String, MappingConfig> rdbMappingTmp = MappingConfigLoader.load(spi.value());
+        // 过滤其他key的配置
+        rdbMappingTmp.forEach((key, mappingConfig) -> {
+            if (mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) {
+                rdbMapping.put(key, mappingConfig);
             }
+        });
+        mappingConfigCache = new HashMap<>();
+        for (MappingConfig mappingConfig : rdbMapping.values()) {
+            mappingConfigCache
+                .put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
+                     + mappingConfig.getDbMapping().getDatabase() + "." + mappingConfig.getDbMapping().getTable(),
+                    mappingConfig);
         }
 
         Map<String, String> properties = configuration.getProperties();

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/MysqlAdapter.java

@@ -1,8 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.rdb.adapters;
-
-import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
-
-@SPI("mysql")
-public class MysqlAdapter extends RdbAdapter {
-}

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/OracleAdapter.java

@@ -1,8 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.rdb.adapters;
-
-import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
-
-@SPI("oracle")
-public class OracleAdapter extends RdbAdapter {
-}

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/PostgresqlAdapter.java

@@ -1,8 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.rdb.adapters;
-
-import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
-
-@SPI("postgresql")
-public class PostgresqlAdapter extends RdbAdapter {
-}

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/adapters/SqlserverAdapter.java

@@ -1,8 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.rdb.adapters;
-
-import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
-import com.alibaba.otter.canal.client.adapter.support.SPI;
-
-@SPI("sqlserver")
-public class SqlserverAdapter extends RdbAdapter {
-}

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java

@@ -12,11 +12,13 @@ import java.util.Set;
  */
 public class MappingConfig {
 
-    private String    dataSourceKey; // 数据源key
+    private String    dataSourceKey;   // 数据源key
 
-    private String    destination;   // canal实例或MQ的topic
+    private String    destination;     // canal实例或MQ的topic
 
-    private DbMapping dbMapping;     // db映射配置
+    private String    outerAdapterKey; // 对应适配器的key
+
+    private DbMapping dbMapping;       // db映射配置
 
     public String getDataSourceKey() {
         return dataSourceKey;
@@ -26,6 +28,14 @@ public class MappingConfig {
         this.dataSourceKey = dataSourceKey;
     }
 
+    public String getOuterAdapterKey() {
+        return outerAdapterKey;
+    }
+
+    public void setOuterAdapterKey(String outerAdapterKey) {
+        this.outerAdapterKey = outerAdapterKey;
+    }
+
     public DbMapping getDbMapping() {
         return dbMapping;
     }

client-adapter/rdb/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter

@@ -1,4 +1 @@
-mysql=com.alibaba.otter.canal.client.adapter.rdb.adapters.MysqlAdapter
-postgresql=com.alibaba.otter.canal.client.adapter.rdb.adapters.PostgresqlAdapter
-oracle=com.alibaba.otter.canal.client.adapter.rdb.adapters.OracleAdapter
-sqlserver=com.alibaba.otter.canal.client.adapter.rdb.adapters.SqlserverAdapter
+rdb=com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter

client-adapter/rdb/src/main/resources/postgresql/mytest_user.yml

@@ -1,15 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-dbMapping:
-  database: mytest
-  table: user
-  targetTable: public.tb_user
-  targetPk:
-    id: id
-  mapAll: true
-#  targetColumns:
-#    id:
-#    name:
-#    role_id:
-#    c_time:
-#    test1:

client-adapter/rdb/src/main/resources/oracle/mytest_user.yml → client-adapter/rdb/src/main/resources/rdb/mytest_user.yml

@@ -1,5 +1,6 @@
 dataSourceKey: defaultDS
 destination: example
+outerAdapterKey: oracle1
 dbMapping:
   database: mytest
   table: user

client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/Common.java

@@ -1,7 +1,6 @@
 package com.alibaba.otter.canal.client.adapter.rdb.test.sync;
 
 import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
-import com.alibaba.otter.canal.client.adapter.rdb.adapters.OracleAdapter;
 import com.alibaba.otter.canal.client.adapter.rdb.test.TestConstant;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
@@ -14,7 +13,8 @@ public class Common {
         DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource);
 
         OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig();
-        outerAdapterConfig.setName("oracle");
+        outerAdapterConfig.setName("rdb");
+        outerAdapterConfig.setKey("oralce1");
         Map<String, String> properties = new HashMap<>();
         properties.put("jdbc.driveClassName", "oracle.jdbc.OracleDriver");
         properties.put("jdbc.url", "jdbc:oracle:thin:@127.0.0.1:49161:XE");
@@ -22,7 +22,7 @@ public class Common {
         properties.put("jdbc.password", "m121212");
         outerAdapterConfig.setProperties(properties);
 
-        RdbAdapter adapter = new OracleAdapter();
+        RdbAdapter adapter = new RdbAdapter();
         adapter.init(outerAdapterConfig);
         return adapter;
     }

client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java

@@ -15,7 +15,7 @@ public class OracleSyncTest {
 
     @Before
     public void init() {
-        AdapterConfigs.put("oracle", "mytest_user.yml");
+        AdapterConfigs.put("rdb", "mytest_user.yml");
         rdbAdapter = Common.init();
     }