diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index a581ec4c..eb3a21a3 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -18,16 +18,21 @@ canal.protocol ${canal_version} - - org.slf4j - slf4j-api - 1.7.12 - joda-time joda-time 2.9.4 + + com.alibaba + druid + 1.1.9 + + + mysql + mysql-connector-java + 5.1.40 + diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java deleted file mode 100644 index 5fe0a377..00000000 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java +++ /dev/null @@ -1,36 +0,0 @@ -package com.alibaba.otter.canal.client.adapter; - -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; -import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.client.adapter.support.SPI; - -/** - * 外部适配器接口 - * - * @author machengyuan 2018-8-18 下午10:14:02 - * @version 1.0.0 - */ -@SPI("logger") -public interface CanalOuterAdapter { - - /** - * 外部适配器初始化接口 - * - * @param configuration 外部适配器配置信息 - */ - void init(CanalOuterAdapterConfiguration configuration); - - /** - * 往适配器中写入数据 - * - * @param dml 数据包 - */ - void writeOut(Dml dml); - - // void writeOut(FlatMessage flatMessage); - - /** - * 外部适配器销毁接口 - */ - void destroy(); -} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java new file mode 100644 index 00000000..286ba398 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java @@ -0,0 +1,68 @@ +package com.alibaba.otter.canal.client.adapter; + +import java.util.List; +import java.util.Map; + +import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +/** + * 外部适配器接口 + * + * @author machengyuan 2018-8-18 下午10:14:02 + * @version 1.0.0 + */ +@SPI("logger") +public interface OuterAdapter { + + /** + * 外部适配器初始化接口 + * + * @param configuration 外部适配器配置信息 + */ + void init(OuterAdapterConfig configuration); + + /** + * 往适配器中同步数据 + * + * @param dml 数据包 + */ + void sync(Dml dml); + + /** + * 外部适配器销毁接口 + */ + void destroy(); + + /** + * Etl操作 + * + * @param task 任务名, 对应配置名 + * @param params etl筛选条件 + */ + default EtlResult etl(String task, List params) { + throw new UnsupportedOperationException("unsupported operation"); + } + + /** + * 计算总数 + * + * @param task 任务名, 对应配置名 + * @return 总数 + */ + default Map count(String task) { + throw new UnsupportedOperationException("unsupported operation"); + } + + /** + * 通过task获取对应的destination + * + * @param task 任务名, 对应配置名 + * @return destination + */ + default String getDestination(String task) { + return null; + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java index 13e51af1..44d29ee9 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java @@ -1,8 +1,28 @@ package com.alibaba.otter.canal.client.adapter.support; -import com.google.common.collect.ArrayListMultimap; -import com.google.common.collect.Multimap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; -public interface AdapterConfigs { - Multimap configs = ArrayListMultimap.create(); +public class AdapterConfigs { + + private static Map> configs = new ConcurrentHashMap<>(); + + public static void put(String key, String value) { + Set values = configs.get(key); + if (values == null) { + values = new LinkedHashSet<>(); + } + values.add(value); + configs.put(key, values); + } + + public static Set get(String key) { + return configs.get(key); + } + + public static void clear() { + configs.clear(); + } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index d885e1cb..08246a42 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -62,6 +62,10 @@ public class CanalClientConfig { return mqTopics; } + public void setMqTopics(List mqTopics) { + this.mqTopics = mqTopics; + } + public Boolean getFlatMessage() { return flatMessage; } @@ -70,10 +74,6 @@ public class CanalClientConfig { this.flatMessage = flatMessage; } - public void setMqTopics(List mqTopics) { - this.mqTopics = mqTopics; - } - public List getCanalInstances() { return canalInstances; } @@ -109,13 +109,13 @@ public class CanalClientConfig { public static class AdapterGroup { - private List outAdapters; + private List outAdapters; - public List getOutAdapters() { + public List getOutAdapters() { return outAdapters; } - public void setOutAdapters(List outAdapters) { + public void setOutAdapters(List outAdapters) { this.outAdapters = outAdapters; } } @@ -155,11 +155,11 @@ public class CanalClientConfig { public static class Group { - private String groupId; + private String groupId; // private List adapters = new ArrayList<>(); - private List outAdapters; + private List outAdapters; public String getGroupId() { return groupId; @@ -169,11 +169,11 @@ public class CanalClientConfig { this.groupId = groupId; } - public List getOutAdapters() { + public List getOutAdapters() { return outAdapters; } - public void setOutAdapters(List outAdapters) { + public void setOutAdapters(List outAdapters) { this.outAdapters = outAdapters; } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/DatasourceConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/DatasourceConfig.java new file mode 100644 index 00000000..cf2ec471 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/DatasourceConfig.java @@ -0,0 +1,75 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import com.alibaba.druid.pool.DruidDataSource; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class DatasourceConfig { + + public final static Map DATA_SOURCES = new ConcurrentHashMap<>(); + + private String driver = "com.mysql.jdbc.Driver"; + private String url; + private String database; + private String type = "mysql"; + private String username; + private String password; + private Integer maxActive = 3; + + public String getDriver() { + return driver; + } + + public void setDriver(String driver) { + this.driver = driver; + } + + public String getUrl() { + return url; + } + + public void setUrl(String url) { + this.url = url; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getUsername() { + return username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Integer getMaxActive() { + return maxActive; + } + + public void setMaxActive(Integer maxActive) { + this.maxActive = maxActive; + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java index d4cbbf7d..8e5af059 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java @@ -14,6 +14,7 @@ public class Dml implements Serializable { private static final long serialVersionUID = 2611556444074013268L; + private String destination; private String database; private String table; private String type; @@ -25,6 +26,14 @@ public class Dml implements Serializable { private List> data; private List> old; + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + public String getDatabase() { return database; } @@ -102,7 +111,8 @@ public class Dml implements Serializable { @Override public String toString() { - return "Dml [database=" + database + ", table=" + table + ", type=" + type + ", es=" + es + ", ts=" + ts - + ", sql=" + sql + ", data=" + data + ", old=" + old + "]"; + return "Dml{" + "destination='" + destination + '\'' + ", database='" + database + '\'' + ", table='" + table + + '\'' + ", type='" + type + '\'' + ", es=" + es + ", ts=" + ts + ", sql='" + sql + '\'' + ", data=" + + data + ", old=" + old + '}'; } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java new file mode 100644 index 00000000..0004e51a --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java @@ -0,0 +1,37 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.io.Serializable; + +public class EtlResult implements Serializable { + private static final long serialVersionUID = 4250522736289866505L; + + private boolean succeeded = false; + + private String resultMessage; + + private String errorMessage; + + public boolean getSucceeded() { + return succeeded; + } + + public void setSucceeded(boolean succeeded) { + this.succeeded = succeeded; + } + + public String getResultMessage() { + return resultMessage; + } + + public void setResultMessage(String resultMessage) { + this.resultMessage = resultMessage; + } + + public String getErrorMessage() { + return errorMessage; + } + + public void setErrorMessage(String errorMessage) { + this.errorMessage = errorMessage; + } +} \ No newline at end of file diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java index 777770ef..d4cb8a0e 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java @@ -2,10 +2,7 @@ package com.alibaba.otter.canal.client.adapter.support; import java.math.BigDecimal; import java.math.BigInteger; -import java.sql.Date; -import java.sql.Time; -import java.sql.Timestamp; -import java.sql.Types; +import java.sql.*; import org.joda.time.DateTime; import org.slf4j.Logger; @@ -21,6 +18,55 @@ public class JdbcTypeUtil { private static Logger logger = LoggerFactory.getLogger(JdbcTypeUtil.class); + public static Object getRSData(ResultSet rs, String columnName, int jdbcType) throws SQLException { + if (jdbcType == Types.BIT || jdbcType == Types.BOOLEAN) { + return rs.getByte(columnName); + } else { + return rs.getObject(columnName); + } + } + + public static Class jdbcType2javaType(int jdbcType) { + switch (jdbcType) { + case Types.BIT: + case Types.BOOLEAN: +// return Boolean.class; + case Types.TINYINT: + return Byte.TYPE; + case Types.SMALLINT: + return Short.class; + case Types.INTEGER: + return Integer.class; + case Types.BIGINT: + return Long.class; + case Types.DECIMAL: + case Types.NUMERIC: + return BigDecimal.class; + case Types.REAL: + return Float.class; + case Types.FLOAT: + case Types.DOUBLE: + return Double.class; + case Types.CHAR: + case Types.VARCHAR: + case Types.LONGVARCHAR: + return String.class; + case Types.BINARY: + case Types.VARBINARY: + case Types.LONGVARBINARY: + case Types.BLOB: + return byte[].class; + case Types.DATE: + return java.sql.Date.class; + case Types.TIME: + return Time.class; + case Types.TIMESTAMP: + return Timestamp.class; + default: + return String.class; + } + } + public static Object typeConvert(String columnName, String value, int sqlType, String mysqlType) { if (value == null || value.equals("")) { return null; diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index 02723b41..826ed917 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -1,11 +1,6 @@ package com.alibaba.otter.canal.client.adapter.support; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -19,7 +14,7 @@ import com.alibaba.otter.canal.protocol.Message; */ public class MessageUtil { - public static void parse4Dml(Message message, Consumer consumer) { + public static void parse4Dml(String destination, Message message, Consumer consumer) { if (message == null) { return; } @@ -42,6 +37,7 @@ public class MessageUtil { CanalEntry.EventType eventType = rowChange.getEventType(); final Dml dml = new Dml(); + dml.setDestination(destination); dml.setDatabase(entry.getHeader().getSchemaName()); dml.setTable(entry.getHeader().getTableName()); dml.setType(eventType.toString()); @@ -87,10 +83,11 @@ public class MessageUtil { Map rowOld = new LinkedHashMap<>(); for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { if (updateSet.contains(column.getName())) { - rowOld.put(column.getName(), JdbcTypeUtil.typeConvert(column.getName(), - column.getValue(), - column.getSqlType(), - column.getMysqlType())); + rowOld.put(column.getName(), + JdbcTypeUtil.typeConvert(column.getName(), + column.getValue(), + column.getSqlType(), + column.getMysqlType())); } } // update操作将记录修改前的值 @@ -110,11 +107,12 @@ public class MessageUtil { } } - public static Dml flatMessage2Dml(FlatMessage flatMessage) { + public static Dml flatMessage2Dml(String destination, FlatMessage flatMessage) { if (flatMessage == null) { return null; } Dml dml = new Dml(); + dml.setDestination(destination); dml.setDatabase(flatMessage.getDatabase()); dml.setTable(flatMessage.getTable()); dml.setType(flatMessage.getType()); diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java similarity index 51% rename from client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java rename to client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java index 790044c8..3aed2f16 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/OuterAdapterConfig.java @@ -1,6 +1,6 @@ package com.alibaba.otter.canal.client.adapter.support; -import java.util.Properties; +import java.util.Map; /** * 外部适配器配置信息类 @@ -8,15 +8,15 @@ import java.util.Properties; * @author machengyuan 2018-8-18 下午10:15:12 * @version 1.0.0 */ -public class CanalOuterAdapterConfiguration { +public class OuterAdapterConfig { - private String name; // 适配器名称, 如: logger, hbase, es + private String name; // 适配器名称, 如: logger, hbase, es - private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址 + private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址 - private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址 + private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址 - private Properties properties; // 其余参数, 可填写适配器中的所需的配置信息 + private Map properties; // 其余参数, 可填写适配器中的所需的配置信息 public String getName() { return name; @@ -34,11 +34,11 @@ public class CanalOuterAdapterConfiguration { this.hosts = hosts; } - public Properties getProperties() { + public Map getProperties() { return properties; } - public void setProperties(Properties properties) { + public void setProperties(Map properties) { this.properties = properties; } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java new file mode 100644 index 00000000..03831f1f --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java @@ -0,0 +1,50 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.io.Serializable; +import java.util.Date; + +public class Result implements Serializable { + + public Integer code = 20000; + public Object data; + public String message; + public Date sysTime; + + public static Result createSuccess(String message) { + Result result = new Result(); + result.setMessage(message); + return result; + } + + public Integer getCode() { + return code; + } + + public void setCode(Integer code) { + this.code = code; + } + + public Object getData() { + return data; + } + + public void setData(Object data) { + this.data = data; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } + + public Date getSysTime() { + return sysTime; + } + + public void setSysTime(Date sysTime) { + this.sysTime = sysTime; + } +} diff --git a/client-adapter/example/pom.xml b/client-adapter/example/pom.xml index 606534e0..bf68d731 100644 --- a/client-adapter/example/pom.xml +++ b/client-adapter/example/pom.xml @@ -55,6 +55,7 @@ maven-jar-plugin + 3.0.2 true diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java index c1f55cd9..bbfd462d 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java @@ -10,7 +10,7 @@ import java.util.concurrent.Future; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.MessageUtil; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -26,10 +26,10 @@ public abstract class AbstractCanalAdapterWorker { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); - protected String canalDestination; // canal实例 - protected List> canalOuterAdapters; // 外部适配器 - protected ExecutorService groupInnerExecutorService; // 组内工作线程池 - protected volatile boolean running = false; // 是否运行中 + protected String canalDestination; // canal实例 + protected List> canalOuterAdapters; // 外部适配器 + protected ExecutorService groupInnerExecutorService; // 组内工作线程池 + protected volatile boolean running = false; // 是否运行中 protected Thread thread = null; protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { @@ -42,21 +42,21 @@ public abstract class AbstractCanalAdapterWorker { protected void writeOut(final Message message) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; futures.add(groupInnerExecutorService.submit(new Callable() { @Override public Boolean call() { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (final CanalOuterAdapter c : adapters) { + for (final OuterAdapter c : adapters) { long begin = System.currentTimeMillis(); - MessageUtil.parse4Dml(message, new MessageUtil.Consumer() { + MessageUtil.parse4Dml(canalDestination, message, new MessageUtil.Consumer() { @Override public void accept(Dml dml) { - c.writeOut(dml); + c.sync(dml); } }); @@ -90,18 +90,18 @@ public abstract class AbstractCanalAdapterWorker { protected void writeOut(final FlatMessage flatMessage) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; futures.add(groupInnerExecutorService.submit(new Callable() { @Override public Boolean call() { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (CanalOuterAdapter c : adapters) { + for (OuterAdapter c : adapters) { long begin = System.currentTimeMillis(); - Dml dml = MessageUtil.flatMessage2Dml(flatMessage); - c.writeOut(dml); + Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage); + c.sync(dml); if (logger.isDebugEnabled()) { logger.debug("{} elapsed time: {}", c.getClass().getName(), @@ -154,8 +154,8 @@ public abstract class AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java index 4429ca80..b2d3e583 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java @@ -4,12 +4,11 @@ import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import org.apache.kafka.clients.consumer.CommitFailedException; import org.apache.kafka.common.errors.WakeupException; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -30,7 +29,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { private boolean flatMessage; public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, - List> canalOuterAdapters, boolean flatMessage){ + List> canalOuterAdapters, boolean flatMessage){ this.canalOuterAdapters = canalOuterAdapters; this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.topic = topic; @@ -81,8 +80,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("topic {} connectors' worker thread dead!", this.topic); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java index 471b6a3d..51b12a0a 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java @@ -13,9 +13,9 @@ import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; /** @@ -33,7 +33,7 @@ public class CanalAdapterLoader { private Map canalMQWorker = new HashMap<>(); - private ExtensionLoader loader; + private ExtensionLoader loader; public CanalAdapterLoader(CanalClientConfig canalClientConfig){ this.canalClientConfig = canalClientConfig; @@ -48,7 +48,7 @@ public class CanalAdapterLoader { throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); } - loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, "" /* + loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class, "" /* * TODO * canalClientConfig * . @@ -68,11 +68,11 @@ public class CanalAdapterLoader { // 初始化canal-client的适配器 if (canalClientConfig.getCanalInstances() != null) { for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new ArrayList<>(); for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { - List canalOutConnectors = new ArrayList<>(); - for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) { + List canalOutConnectors = new ArrayList<>(); + for (OuterAdapterConfig c : connectorGroup.getOutAdapters()) { loadConnector(c, canalOutConnectors); } canalOuterAdapterGroups.add(canalOutConnectors); @@ -93,11 +93,11 @@ public class CanalAdapterLoader { if (canalClientConfig.getMqTopics() != null) { for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { for (CanalClientConfig.Group group : topic.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new ArrayList<>(); - List canalOuterAdapters = new ArrayList<>(); + List canalOuterAdapters = new ArrayList<>(); - for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) { + for (OuterAdapterConfig config : group.getOutAdapters()) { loadConnector(config, canalOuterAdapters); } canalOuterAdapterGroups.add(canalOuterAdapters); @@ -125,9 +125,9 @@ public class CanalAdapterLoader { } } - private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { + private void loadConnector(OuterAdapterConfig config, List canalOutConnectors) { try { - CanalOuterAdapter adapter = loader.getExtension(config.getName()); + OuterAdapter adapter = loader.getExtension(config.getName()); ClassLoader cl = Thread.currentThread().getContextClassLoader(); // 替换ClassLoader Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java index f0f674db..9c126297 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java @@ -8,7 +8,7 @@ import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.CommitFailedException; import org.apache.kafka.common.errors.WakeupException; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; import com.alibaba.otter.canal.protocol.Message; @@ -26,7 +26,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { private String topic; public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, - List> canalOuterAdapters){ + List> canalOuterAdapters){ logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); this.canalOuterAdapters = canalOuterAdapters; this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java index e341c6a3..e72ed2d5 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java @@ -6,7 +6,7 @@ import java.util.concurrent.Executors; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.protocol.Message; @@ -31,7 +31,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, SocketAddress address, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -46,7 +46,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, String zookeeperHosts, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -96,8 +96,8 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("destination {} adapters' worker thread dead!", canalDestination); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index 866bd31f..a48d7401 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -84,25 +84,6 @@ - - org.apache.maven.plugins - maven-assembly-plugin - 2.4 - - - jar-with-dependencies - - - - - make-assembly - package - - single - - - - diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index d75cea67..ababeb6e 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -2,21 +2,29 @@ package com.alibaba.otter.canal.client.adapter.hbase; import java.io.IOException; import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; +import javax.sql.DataSource; + import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.Connection; -import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.*; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader; +import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseEtlService; import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService; -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; -import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.client.adapter.support.SPI; +import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate; +import com.alibaba.otter.canal.client.adapter.support.*; /** * HBase外部适配器 @@ -25,74 +33,128 @@ import com.alibaba.otter.canal.client.adapter.support.SPI; * @version 1.0.0 */ @SPI("hbase") -public class HbaseAdapter implements CanalOuterAdapter { +public class HbaseAdapter implements OuterAdapter { - private static volatile Map mappingConfigCache = null; + private static Logger logger = LoggerFactory.getLogger(HbaseAdapter.class); + + private static volatile Map hbaseMapping = null; // 文件名对应配置 + private static volatile Map mappingConfigCache = null; // 库名-表名对应配置 private Connection conn; private HbaseSyncService hbaseSyncService; + private HbaseTemplate hbaseTemplate; @Override - public void init(CanalOuterAdapterConfiguration configuration) { + public void init(OuterAdapterConfig configuration) { try { if (mappingConfigCache == null) { synchronized (MappingConfig.class) { if (mappingConfigCache == null) { - Map hbaseMapping = MappingConfigLoader.load(); + hbaseMapping = MappingConfigLoader.load(); mappingConfigCache = new HashMap<>(); for (MappingConfig mappingConfig : hbaseMapping.values()) { - mappingConfigCache.put(mappingConfig.getHbaseOrm().getDatabase() + "-" - + mappingConfig.getHbaseOrm().getTable(), mappingConfig); + mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getHbaseOrm().getDestination()) + + "." + mappingConfig.getHbaseOrm().getDatabase() + "." + + mappingConfig.getHbaseOrm().getTable(), + mappingConfig); } } } } - String hosts = configuration.getZkHosts(); - if (StringUtils.isEmpty(hosts)) { - hosts = configuration.getHosts(); - } - if (StringUtils.isEmpty(hosts)) { - throw new RuntimeException("Empty zookeeper hosts"); - } - String[] zkHosts = StringUtils.split(hosts, ","); - int zkPort = 0; - StringBuilder hostsWithoutPort = new StringBuilder(); - for (String host : zkHosts) { - int i = host.indexOf(":"); - hostsWithoutPort.append(host, 0, i); - hostsWithoutPort.append(","); - if (zkPort == 0) zkPort = Integer.parseInt(host.substring(i + 1)); - } - hostsWithoutPort.deleteCharAt(hostsWithoutPort.length() - 1); - - String znode = configuration.getProperties().getProperty("znodeParent"); - if (StringUtils.isEmpty(znode)) { - znode = "/hbase"; - } + Map propertites = configuration.getProperties(); Configuration hbaseConfig = HBaseConfiguration.create(); - hbaseConfig.set("hbase.zookeeper.quorum", hostsWithoutPort.toString()); - hbaseConfig.set("hbase.zookeeper.property.clientPort", Integer.toString(zkPort)); - hbaseConfig.set("zookeeper.znode.parent", znode); + propertites.forEach(hbaseConfig::set); conn = ConnectionFactory.createConnection(hbaseConfig); - hbaseSyncService = new HbaseSyncService(conn); + hbaseTemplate = new HbaseTemplate(conn); + hbaseSyncService = new HbaseSyncService(hbaseTemplate); } catch (Exception e) { throw new RuntimeException(e); } } @Override - public void writeOut(Dml dml) { + public void sync(Dml dml) { if (dml == null) { return; } + String destination = StringUtils.trimToEmpty(dml.getDestination()); String database = dml.getDatabase(); String table = dml.getTable(); - MappingConfig config = mappingConfigCache.get(database + "-" + table); + MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); hbaseSyncService.sync(config, dml); } + @Override + public EtlResult etl(String task, List params) { + EtlResult etlResult = new EtlResult(); + MappingConfig config = hbaseMapping.get(task); + if (config != null) { + DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); + if (dataSource != null) { + return HbaseEtlService.importData(dataSource, hbaseTemplate, config, params); + } else { + etlResult.setSucceeded(false); + etlResult.setErrorMessage("DataSource not found"); + return etlResult; + } + } else { + DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(task); + if (dataSource != null) { + StringBuilder resultMsg = new StringBuilder(); + boolean resSucc = true; + // ds不为空说明传入的是datasourceKey + for (MappingConfig configTmp : hbaseMapping.values()) { + // 取所有的datasourceKey为task的配置 + if (configTmp.getDataSourceKey().equals(task)) { + EtlResult etlRes = HbaseEtlService.importData(dataSource, hbaseTemplate, configTmp, params); + if (!etlRes.getSucceeded()) { + resSucc = false; + resultMsg.append(etlRes.getErrorMessage()).append("\n"); + } else { + resultMsg.append(etlRes.getResultMessage()).append("\n"); + } + } + } + if (resultMsg.length() > 0) { + etlResult.setSucceeded(resSucc); + if (resSucc) { + etlResult.setResultMessage(resultMsg.toString()); + } else { + etlResult.setErrorMessage(resultMsg.toString()); + } + return etlResult; + } + } + } + etlResult.setSucceeded(false); + etlResult.setErrorMessage("Task not found"); + return etlResult; + } + + @Override + public Map count(String task) { + MappingConfig config = hbaseMapping.get(task); + String hbaseTable = config.getHbaseOrm().getHbaseTable(); + long rowCount = 0L; + try { + HTable table = (HTable) conn.getTable(TableName.valueOf(hbaseTable)); + Scan scan = new Scan(); + scan.setFilter(new FirstKeyOnlyFilter()); + ResultScanner resultScanner = table.getScanner(scan); + for (Result result : resultScanner) { + rowCount += result.size(); + } + } catch (IOException e) { + logger.error(e.getMessage(), e); + } + Map res = new LinkedHashMap<>(); + res.put("hbaseTable", hbaseTable); + res.put("count", rowCount); + return res; + } + @Override public void destroy() { if (conn != null) { @@ -103,4 +165,13 @@ public class HbaseAdapter implements CanalOuterAdapter { } } } + + @Override + public String getDestination(String task) { + MappingConfig config = hbaseMapping.get(task); + if (config != null && config.getHbaseOrm() != null) { + return config.getHbaseOrm().getDestination(); + } + return null; + } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java index 6d70bdde..a5ea5707 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java @@ -13,9 +13,18 @@ import java.util.Set; * @version 1.0.0 */ public class MappingConfig { + private String dataSourceKey; private HbaseOrm hbaseOrm; + public String getDataSourceKey() { + return dataSourceKey; + } + + public void setDataSourceKey(String dataSourceKey) { + this.dataSourceKey = dataSourceKey; + } + public HbaseOrm getHbaseOrm() { return hbaseOrm; } @@ -137,6 +146,7 @@ public class MappingConfig { public static class HbaseOrm { private Mode mode = Mode.STRING; + private String destination; private String database; private String table; private String hbaseTable; @@ -161,6 +171,14 @@ public class MappingConfig { this.mode = mode; } + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + public String getDatabase() { return database; } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java index 5922d63e..0fb727d8 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java @@ -37,7 +37,7 @@ public class MappingConfigLoader { Map result = new LinkedHashMap<>(); - Collection configs = AdapterConfigs.configs.get("hbase"); + Collection configs = AdapterConfigs.get("hbase"); for (String c : configs) { if (c == null) { continue; @@ -72,6 +72,7 @@ public class MappingConfigLoader { String[] dbTable; if (dsKey == null) { dbTable = srcMeta.split("\\."); + } else { dbTable = srcMeta.split("@")[0].split("\\."); } @@ -84,7 +85,7 @@ public class MappingConfigLoader { hbaseOrm.setAutoCreateTable(true); hbaseOrm.setDatabase(dbTable[0]); hbaseOrm.setTable(dbTable[1]); - hbaseOrm.setMode(MappingConfig.Mode.STRING); + hbaseOrm.setMode(MappingConfig.Mode.PHOENIX); hbaseOrm.setRowKey(rowKey); // 有定义rowKey if (rowKey != null) { @@ -94,6 +95,7 @@ public class MappingConfigLoader { hbaseOrm.setRowKeyColumn(columnItem); } config.setHbaseOrm(hbaseOrm); + config.setDataSourceKey(dsKey); } else { throw new RuntimeException(String.format("配置项[%s]内容为空, 或格式不符合database.table", c)); diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java new file mode 100644 index 00000000..ee110755 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java @@ -0,0 +1,353 @@ +package com.alibaba.otter.canal.client.adapter.hbase.service; + +import java.sql.*; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; + +import javax.sql.DataSource; + +import org.apache.hadoop.hbase.util.Bytes; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; +import com.alibaba.otter.canal.client.adapter.hbase.support.*; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil; +import com.google.common.base.Joiner; + +public class HbaseEtlService { + + private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class); + + public static Object sqlRS(DataSource ds, String sql, Function fun) throws SQLException { + Connection conn = null; + Statement stmt = null; + ResultSet rs = null; + try { + conn = ds.getConnection(); + stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); + stmt.setFetchSize(Integer.MIN_VALUE); + rs = stmt.executeQuery(sql); + return fun.apply(rs); + } finally { + if (rs != null) { + try { + rs.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + if (stmt != null) { + try { + stmt.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + if (conn != null) { + try { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } + rs = null; + stmt = null; + conn = null; + } + } + + public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) { + try { + // 判断hbase表是否存在，不存在则建表 + MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm(); + if (!hbaseTemplate.tableExists(hbaseOrm.getHbaseTable())) { + hbaseTemplate.createTable(hbaseOrm.getHbaseTable(), hbaseOrm.getFamily()); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + throw new RuntimeException(e); + } + } + + public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config, + List params) { + EtlResult etlResult = new EtlResult(); + AtomicLong successCount = new AtomicLong(); + List errMsg = new ArrayList<>(); + String hbaseTable = ""; + try { + if (config == null) { + logger.error("Config is null!"); + etlResult.setSucceeded(false); + etlResult.setErrorMessage("Config is null!"); + return etlResult; + } + MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm(); + hbaseTable = hbaseOrm.getHbaseTable(); + + long start = System.currentTimeMillis(); + + if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) { + logger.info(hbaseOrm.getHbaseTable() + " rebuild is starting!"); + // 如果表存在则删除 + if (hbaseTemplate.tableExists(hbaseOrm.getHbaseTable())) { + hbaseTemplate.disableTable(hbaseOrm.getHbaseTable()); + hbaseTemplate.deleteTable(hbaseOrm.getHbaseTable()); + } + params = null; + } else { + logger.info(hbaseOrm.getHbaseTable() + " etl is starting!"); + } + createTable(hbaseTemplate, config); + + // 拼接sql + String sql = "SELECT * FROM " + config.getHbaseOrm().getDatabase() + "." + hbaseOrm.getTable(); + + // 拼接条件 + if (params != null && params.size() == 1 && hbaseOrm.getEtlCondition() == null) { + AtomicBoolean stExists = new AtomicBoolean(false); + // 验证是否有SYS_TIME字段 + sqlRS(ds, sql, rs -> { + try { + ResultSetMetaData rsmd = rs.getMetaData(); + int cnt = rsmd.getColumnCount(); + for (int i = 1; i <= cnt; i++) { + String columnName = rsmd.getColumnName(i); + if ("SYS_TIME".equalsIgnoreCase(columnName)) { + stExists.set(true); + break; + } + } + } catch (Exception e) { + // ignore + } + return null; + }); + if (stExists.get()) { + sql += " WHERE SYS_TIME >= '" + params.get(0) + "' "; + } + } else if (hbaseOrm.getEtlCondition() != null && params != null) { + String etlCondition = hbaseOrm.getEtlCondition(); + int size = params.size(); + for (int i = 0; i < size; i++) { + etlCondition = etlCondition.replace("{" + i + "}", params.get(i)); + } + + sql += " " + etlCondition; + } + + // 获取总数 + String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; + long cnt = (Long) sqlRS(ds, countSql, rs -> { + Long count = null; + try { + if (rs.next()) { + count = ((Number) rs.getObject(1)).longValue(); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + return count == null ? 0 : count; + }); + + // 当大于1万条记录时开启多线程 + if (cnt >= 10000) { + int threadCount = 3; + long perThreadCnt = cnt / threadCount; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + List> futures = new ArrayList<>(threadCount); + for (int i = 0; i < threadCount; i++) { + long offset = i * perThreadCnt; + Long size = null; + if (i != threadCount - 1) { + size = perThreadCnt; + } + String sqlFinal; + if (size != null) { + sqlFinal = sql + " LIMIT " + offset + "," + size; + } else { + sqlFinal = sql + " LIMIT " + offset + "," + cnt; + } + Future future = executor + .submit(() -> executeSqlImport(ds, sqlFinal, hbaseOrm, hbaseTemplate, successCount, errMsg)); + futures.add(future); + } + + for (Future future : futures) { + future.get(); + } + + executor.shutdown(); + } else { + executeSqlImport(ds, sql, hbaseOrm, hbaseTemplate, successCount, errMsg); + } + + logger.info( + hbaseOrm.getHbaseTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!"); + + etlResult.setResultMessage("导入HBase表 " + hbaseOrm.getHbaseTable() + " 数据：" + successCount.get() + " 条"); + } catch (Exception e) { + logger.error(e.getMessage(), e); + errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); + } + + if (errMsg.isEmpty()) { + etlResult.setSucceeded(true); + } else { + etlResult.setErrorMessage(Joiner.on("\n").join(errMsg)); + } + return etlResult; + } + + private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseOrm hbaseOrm, + HbaseTemplate hbaseTemplate, AtomicLong successCount, List errMsg) { + try { + sqlRS(ds, sql, rs -> { + int i = 1; + + try { + boolean complete = false; + List rows = new ArrayList<>(); + String[] rowKeyColumns = null; + if (hbaseOrm.getRowKey() != null) { + rowKeyColumns = hbaseOrm.getRowKey().trim().split(","); + } + while (rs.next()) { + int cc = rs.getMetaData().getColumnCount(); + int[] jdbcTypes = new int[cc]; + Class[] classes = new Class[cc]; + for (int j = 1; j <= cc; j++) { + int jdbcType = rs.getMetaData().getColumnType(j); + jdbcTypes[j - 1] = jdbcType; + classes[j - 1] = JdbcTypeUtil.jdbcType2javaType(jdbcType); + } + HRow row = new HRow(); + + if (rowKeyColumns != null) { + // 取rowKey字段拼接 + StringBuilder rowKeyVale = new StringBuilder(); + for (String rowKeyColumnName : rowKeyColumns) { + Object obj = rs.getObject(rowKeyColumnName); + if (obj != null) { + rowKeyVale.append(obj.toString()); + } + rowKeyVale.append("|"); + } + int len = rowKeyVale.length(); + if (len > 0) { + rowKeyVale.delete(len - 1, len); + } + row.setRowKey(Bytes.toBytes(rowKeyVale.toString())); + } + + for (int j = 1; j <= cc; j++) { + String columnName = rs.getMetaData().getColumnName(j); + + Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]); + if (val == null) { + continue; + } + + MappingConfig.ColumnItem columnItem = hbaseOrm.getColumnItems().get(columnName); + // 没有配置映射 + if (columnItem == null) { + String family = hbaseOrm.getFamily(); + String qualifile = columnName; + if (hbaseOrm.isUppercaseQualifier()) { + qualifile = qualifile.toUpperCase(); + } + if (MappingConfig.Mode.STRING == hbaseOrm.getMode()) { + if (hbaseOrm.getRowKey() == null && j == 1) { + row.setRowKey(Bytes.toBytes(val.toString())); + } else { + row.addCell(family, qualifile, Bytes.toBytes(val.toString())); + } + } else if (MappingConfig.Mode.NATIVE == hbaseOrm.getMode()) { + Type type = Type.getType(classes[j - 1]); + if (hbaseOrm.getRowKey() == null && j == 1) { + row.setRowKey(TypeUtil.toBytes(val, type)); + } else { + row.addCell(family, qualifile, TypeUtil.toBytes(val, type)); + } + } else if (MappingConfig.Mode.PHOENIX == hbaseOrm.getMode()) { + PhType phType = PhType.getType(classes[j - 1]); + if (hbaseOrm.getRowKey() == null && j == 1) { + row.setRowKey(PhTypeUtil.toBytes(val, phType)); + } else { + row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType)); + } + } + } else { + // 如果不需要类型转换 + if (columnItem.getType() == null || "".equals(columnItem.getType())) { + if (val instanceof java.sql.Date) { + SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd"); + val = dateFmt.format((Date) val); + } else if (val instanceof Timestamp) { + SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + val = datetimeFmt.format((Date) val); + } + + byte[] valBytes = Bytes.toBytes(val.toString()); + if (columnItem.isRowKey()) { + row.setRowKey(valBytes); + } else { + row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); + } + } else { + PhType phType = PhType.getType(columnItem.getType()); + if (columnItem.isRowKey()) { + row.setRowKey(PhTypeUtil.toBytes(val, phType)); + } else { + row.addCell(columnItem.getFamily(), + columnItem.getQualifier(), + PhTypeUtil.toBytes(val, phType)); + } + } + } + } + + if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空"); + + rows.add(row); + complete = false; + if (i % hbaseOrm.getCommitBatch() == 0 && !rows.isEmpty()) { + hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows); + rows.clear(); + complete = true; + } + i++; + successCount.incrementAndGet(); + if (logger.isDebugEnabled()) { + logger.debug("successful import count:" + successCount.get()); + } + } + + if (!complete && !rows.isEmpty()) { + hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows); + } + + } catch (Exception e) { + logger.error(hbaseOrm.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e); + errMsg.add(hbaseOrm.getHbaseTable() + " etl failed! ==>" + e.getMessage()); + // throw new RuntimeException(e); + } + return i; + }); + return true; + } catch (Exception e) { + logger.error(e.getMessage(), e); + return false; + } + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java index 7e69f9d0..1ff406fd 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java @@ -1,23 +1,13 @@ package com.alibaba.otter.canal.client.adapter.hbase.service; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; -import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.util.Bytes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; -import com.alibaba.otter.canal.client.adapter.hbase.support.HRow; -import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate; -import com.alibaba.otter.canal.client.adapter.hbase.support.PhType; -import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil; -import com.alibaba.otter.canal.client.adapter.hbase.support.Type; -import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil; +import com.alibaba.otter.canal.client.adapter.hbase.support.*; import com.alibaba.otter.canal.client.adapter.support.Dml; /** @@ -32,8 +22,8 @@ public class HbaseSyncService { private HbaseTemplate hbaseTemplate; // HBase操作模板 - public HbaseSyncService(Connection conn){ - hbaseTemplate = new HbaseTemplate(conn); + public HbaseSyncService(HbaseTemplate hbaseTemplate){ + this.hbaseTemplate = hbaseTemplate; } public void sync(MappingConfig config, Dml dml) { @@ -375,7 +365,8 @@ public class HbaseSyncService { * @param value 值 * @return 复合字段rowKey */ - private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm, Object value) { + private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm, + Object value) { if (value == null) { return null; } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java index 07aad00f..f90e5bd8 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java @@ -92,6 +92,7 @@ public class TypeUtil { return b; } + @SuppressWarnings("unchecked") public static T toObject(byte[] bytes, Class clazz) { if (bytes == null) { return null; @@ -132,10 +133,10 @@ public class TypeUtil { } else { throw new IllegalArgumentException("mismatch class type"); } - // noinspection unchecked return (T) res; } + @SuppressWarnings("unchecked") public static T toObject(byte[] bytes, Type type) { if (bytes == null) { return null; @@ -182,7 +183,6 @@ public class TypeUtil { } else { throw new IllegalArgumentException("mismatch class type"); } - // noinspection unchecked return (T) res; } } diff --git a/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter similarity index 100% rename from client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter rename to client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter diff --git a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml index 575a1bf5..88cd99e2 100644 --- a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml +++ b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml @@ -1,5 +1,7 @@ +dataSourceKey: defaultDS hbaseOrm: mode: PHOENIX #NATIVE #STRING + destination: example database: mytest # 数据库名 table: person2 # 数据库表名 hbaseTable: MYTEST.PERSON2 # HBase表名 diff --git a/client-adapter/launcher/assembly.xml b/client-adapter/launcher/assembly.xml new file mode 100644 index 00000000..7bb12314 --- /dev/null +++ b/client-adapter/launcher/assembly.xml @@ -0,0 +1,34 @@ + + canal-adapter + + tar.gz + + + + ${project.build.directory} + / + + *.jar + + + + + ${project.build.directory}/lib + /lib/ + + *.jar + + + + + ${project.build.directory}/config + /config/ + + ** + + + + / + \ No newline at end of file diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 76480a70..cb8aee72 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -50,6 +50,23 @@ spring-boot-configuration-processor true + + org.apache.curator + curator-recipes + 2.10.0 + + + + org.apache.rocketmq + rocketmq-client + 4.3.0 + + + + org.apache.kafka + kafka-clients + 1.1.1 + @@ -81,6 +98,7 @@ + canal-adapter-launcher org.springframework.boot @@ -132,6 +150,24 @@ + + maven-assembly-plugin + + false + + assembly.xml + + + + + make-assembly + package + + single + + + + diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java index 4b0035d8..56bbde8e 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java @@ -1,56 +1,12 @@ package com.alibaba.otter.canal.adapter.launcher; -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; -import javax.annotation.Resource; - -import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterLoader; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; - @SpringBootApplication public class CanalAdapterApplication { - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterApplication.class); - - private static CanalAdapterLoader adapterLoader; public static void main(String[] args) { new SpringApplicationBuilder(CanalAdapterApplication.class).run(args); } - - @Resource - private CanalClientConfig canalClientConf; - - @PostConstruct - public void init() { - if (adapterLoader == null) { - try { - logger.info("## start the canal client adapters."); - adapterLoader = new CanalAdapterLoader(canalClientConf); - adapterLoader.init(); - logger.info("## the canal client adapters are running now ......"); - } catch (Throwable e) { - logger.error("## something goes wrong when starting up the canal client adapters:", e); - System.exit(0); - } - } - } - - @PreDestroy - public void destroy() { - try { - logger.info("## stop the canal client adapters"); - if (adapterLoader != null) { - adapterLoader.destroy(); - } - } catch (Throwable e) { - logger.warn("## something goes wrong when stopping canal client adapters:", e); - } finally { - logger.info("## canal client adapters are down."); - } - } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java new file mode 100644 index 00000000..67d0ac41 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java @@ -0,0 +1,114 @@ +package com.alibaba.otter.canal.adapter.launcher.common; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.locks.InterProcessMutex; +import org.springframework.stereotype.Component; + +import com.alibaba.otter.canal.adapter.launcher.config.CuratorClient; + +@Component +public class EtlLock { + + private static final Map LOCAL_LOCK = new ConcurrentHashMap<>(); + + private static final Map DISTRIBUTED_LOCK = new ConcurrentHashMap<>(); + + private static Mode mode = Mode.LOCAL; + + @Resource + private CuratorClient curatorClient; + + @PostConstruct + public void init() { + CuratorFramework curator = curatorClient.getCurator(); + if (curator != null) { + mode = Mode.DISTRIBUTED; + } else { + mode = Mode.LOCAL; + } + } + + private ReentrantLock getLock(String key) { + ReentrantLock lock = LOCAL_LOCK.get(key); + if (lock == null) { + synchronized (EtlLock.class) { + lock = LOCAL_LOCK.get(key); + if (lock == null) { + lock = new ReentrantLock(); + LOCAL_LOCK.put(key, lock); + } + } + } + return lock; + } + + private InterProcessMutex getRemoteLock(String key) { + InterProcessMutex lock = DISTRIBUTED_LOCK.get(key); + if (lock == null) { + synchronized (EtlLock.class) { + lock = DISTRIBUTED_LOCK.get(key); + if (lock == null) { + lock = new InterProcessMutex(curatorClient.getCurator(), key); + DISTRIBUTED_LOCK.put(key, lock); + } + } + } + return lock; + } + + public void lock(String key) throws Exception { + if (mode == Mode.LOCAL) { + getLock(key).lock(); + } else { + InterProcessMutex lock = getRemoteLock(key); + lock.acquire(); + } + } + + public boolean tryLock(String key, long timeout, TimeUnit unit) { + try { + if (mode == Mode.LOCAL) { + return getLock(key).tryLock(timeout, unit); + } else { + InterProcessMutex lock = getRemoteLock(key); + return lock.acquire(timeout, unit); + } + } catch (Exception e) { + return false; + } + } + + public boolean tryLock(String key) { + try { + if (mode == Mode.LOCAL) { + return getLock(key).tryLock(); + } else { + InterProcessMutex lock = getRemoteLock(key); + return lock.acquire(500, TimeUnit.MILLISECONDS); + } + } catch (Exception e) { + return false; + } + } + + public void unlock(String key) { + if (mode == Mode.LOCAL) { + getLock(key).unlock(); + } else { + InterProcessMutex lock = getRemoteLock(key); + try { + lock.release(); + } catch (Exception e) { + // ignore + } + } + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/Mode.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/Mode.java new file mode 100644 index 00000000..ac73b85e --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/Mode.java @@ -0,0 +1,6 @@ +package com.alibaba.otter.canal.adapter.launcher.common; + +public enum Mode { + LOCAL, // 本地模式 + DISTRIBUTED // 分布式 +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java new file mode 100644 index 00000000..d96c753e --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java @@ -0,0 +1,208 @@ +package com.alibaba.otter.canal.adapter.launcher.common; + +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.cache.NodeCache; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.data.Stat; +import org.springframework.stereotype.Component; + +import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; +import com.alibaba.otter.canal.adapter.launcher.config.CuratorClient; +import com.alibaba.otter.canal.common.utils.BooleanMutex; + +@Component +public class SyncSwitch { + + private static final String SYN_SWITCH_ZK_NODE = "/sync-switch/"; + + private static final Map LOCAL_LOCK = new ConcurrentHashMap<>(); + + private static final Map DISTRIBUTED_LOCK = new ConcurrentHashMap<>(); + + private static Mode mode = Mode.LOCAL; + + @Resource + private AdapterCanalConfig adapterCanalConfig; + @Resource + private CuratorClient curatorClient; + + @PostConstruct + public void init() { + CuratorFramework curator = curatorClient.getCurator(); + if (curator != null) { + mode = Mode.DISTRIBUTED; + DISTRIBUTED_LOCK.clear(); + for (String destination : adapterCanalConfig.DESTINATIONS) { + // 对应每个destination注册锁 + BooleanMutex mutex = new BooleanMutex(true); + initMutex(curator, destination, mutex); + DISTRIBUTED_LOCK.put(destination, mutex); + startListen(destination, mutex); + } + } else { + mode = Mode.LOCAL; + LOCAL_LOCK.clear(); + for (String destination : adapterCanalConfig.DESTINATIONS) { + // 对应每个destination注册锁 + LOCAL_LOCK.put(destination, new BooleanMutex(true)); + } + } + } + + private synchronized void startListen(String destination, BooleanMutex mutex) { + try { + String path = SYN_SWITCH_ZK_NODE + destination; + CuratorFramework curator = curatorClient.getCurator(); + final NodeCache nodeCache = new NodeCache(curator, path); + nodeCache.start(); + nodeCache.getListenable().addListener(() -> initMutex(curator, destination, mutex)); + } catch (Exception e) { + throw new RuntimeException(e.getMessage()); + } + } + + private synchronized void initMutex(CuratorFramework curator, String destination, BooleanMutex mutex) { + try { + String path = SYN_SWITCH_ZK_NODE + destination; + Stat stat = curator.checkExists().forPath(path); + if (stat == null) { + if (!mutex.state()) { + mutex.set(true); + } + } else { + String data = new String(curator.getData().forPath(path), StandardCharsets.UTF_8); + if ("on".equals(data)) { + if (!mutex.state()) { + mutex.set(true); + } + } else { + if (mutex.state()) { + mutex.set(false); + } + } + } + } catch (Exception e) { + throw new RuntimeException(e.getMessage()); + } + } + + public synchronized void off(String destination) { + if (mode == Mode.LOCAL) { + BooleanMutex mutex = LOCAL_LOCK.get(destination); + if (mutex != null && mutex.state()) { + mutex.set(false); + } + } else { + try { + String path = SYN_SWITCH_ZK_NODE + destination; + try { + curatorClient.getCurator() + .create() + .creatingParentContainersIfNeeded() + .withMode(CreateMode.PERSISTENT) + .forPath(path, "off".getBytes(StandardCharsets.UTF_8)); + } catch (Exception e) { + curatorClient.getCurator().setData().forPath(path, "off".getBytes(StandardCharsets.UTF_8)); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + } + + public synchronized void on(String destination) { + if (mode == Mode.LOCAL) { + BooleanMutex mutex = LOCAL_LOCK.get(destination); + if (mutex != null && !mutex.state()) { + mutex.set(true); + } + } else { + try { + String path = SYN_SWITCH_ZK_NODE + destination; + try { + curatorClient.getCurator() + .create() + .creatingParentContainersIfNeeded() + .withMode(CreateMode.PERSISTENT) + .forPath(path, "on".getBytes(StandardCharsets.UTF_8)); + } catch (Exception e) { + curatorClient.getCurator().setData().forPath(path, "on".getBytes(StandardCharsets.UTF_8)); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + } + + public synchronized void release(String destination) { + if (mode == Mode.LOCAL) { + BooleanMutex mutex = LOCAL_LOCK.get(destination); + if (mutex != null && !mutex.state()) { + mutex.set(true); + } + } + if (mode == Mode.DISTRIBUTED) { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null && !mutex.state()) { + mutex.set(true); + } + } + } + + public Boolean status(String destination) { + if (mode == Mode.LOCAL) { + BooleanMutex mutex = LOCAL_LOCK.get(destination); + if (mutex != null) { + return mutex.state(); + } else { + return null; + } + } else { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null) { + return mutex.state(); + } else { + return null; + } + } + } + + public void get(String destination) throws InterruptedException { + if (mode == Mode.LOCAL) { + BooleanMutex mutex = LOCAL_LOCK.get(destination); + if (mutex != null) { + mutex.get(); + } + } else { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null) { + mutex.get(); + } + } + } + + public void get(String destination, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { + if (mode == Mode.LOCAL) { + BooleanMutex mutex = LOCAL_LOCK.get(destination); + if (mutex != null) { + mutex.get(timeout, unit); + } + } else { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null) { + mutex.get(timeout, unit); + } + } + } + + +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java index 5b232c36..18a12d7c 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java @@ -1,10 +1,45 @@ package com.alibaba.otter.canal.adapter.launcher.config; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; -@ConfigurationProperties(prefix = "canal.conf") +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; + @Component +@ConfigurationProperties(prefix = "canal.conf") public class AdapterCanalConfig extends CanalClientConfig { + + public final Set DESTINATIONS = new LinkedHashSet<>(); + + @Override + public void setCanalInstances(List canalInstances) { + super.setCanalInstances(canalInstances); + + if (canalInstances != null) { + synchronized (DESTINATIONS) { + DESTINATIONS.clear(); + for (CanalInstance canalInstance : canalInstances) { + DESTINATIONS.add(canalInstance.getInstance()); + } + } + } + } + + @Override + public void setMqTopics(List mqTopics) { + super.setMqTopics(mqTopics); + + if (mqTopics != null) { + synchronized (DESTINATIONS) { + DESTINATIONS.clear(); + for (MQTopic mqTopic : mqTopics) { + DESTINATIONS.add(mqTopic.getTopic()); + } + } + } + } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java new file mode 100644 index 00000000..da37804f --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java @@ -0,0 +1,80 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import java.sql.SQLException; +import java.util.List; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; + +@Component +@ConfigurationProperties(prefix = "adapter.conf") +public class AdapterConfig { + + private static Logger logger = LoggerFactory.getLogger(AdapterConfig.class); + + private Map datasourceConfigs; + + private List adapterConfigs; + + public List getAdapterConfigs() { + return adapterConfigs; + } + + public Map getDatasourceConfigs() { + return datasourceConfigs; + } + + public void setDatasourceConfigs(Map datasourceConfigs) { + this.datasourceConfigs = datasourceConfigs; + + if (datasourceConfigs != null) { + for (Map.Entry entry : datasourceConfigs.entrySet()) { + DatasourceConfig datasourceConfig = entry.getValue(); + // 加载数据源连接池 + DruidDataSource ds = new DruidDataSource(); + ds.setDriverClassName(datasourceConfig.getDriver()); + ds.setUrl(datasourceConfig.getUrl()); + ds.setUsername(datasourceConfig.getUsername()); + ds.setPassword(datasourceConfig.getPassword()); + ds.setInitialSize(1); + ds.setMinIdle(1); + ds.setMaxActive(datasourceConfig.getMaxActive()); + ds.setMaxWait(60000); + ds.setTimeBetweenEvictionRunsMillis(60000); + ds.setMinEvictableIdleTimeMillis(300000); + ds.setPoolPreparedStatements(false); + ds.setMaxPoolPreparedStatementPerConnectionSize(20); + ds.setValidationQuery("select 1"); + try { + ds.init(); + } catch (SQLException e) { + logger.error("#Failed to initial datasource: " + datasourceConfig.getUrl(), e); + } + DatasourceConfig.DATA_SOURCES.put(entry.getKey(), ds); + } + } + } + + public void setAdapterConfigs(List adapterConfigs) { + this.adapterConfigs = adapterConfigs; + + if (adapterConfigs != null) { + AdapterConfigs.clear(); + for (String adapterConfig : adapterConfigs) { + int idx = adapterConfig.indexOf("/"); + if (idx > -1) { + String type = adapterConfig.substring(0, idx); + String ymlFile = adapterConfig.substring(idx + 1); + AdapterConfigs.put(type, ymlFile); + } + } + } + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java deleted file mode 100644 index 474d43af..00000000 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java +++ /dev/null @@ -1,35 +0,0 @@ -package com.alibaba.otter.canal.adapter.launcher.config; - -import java.util.List; - -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.stereotype.Component; - -import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; - -@Component -@ConfigurationProperties(prefix = "adapter.conf.path") -public class AdapterConfigPath { - - private List adapterConfigs; - - public List getAdapterConfigs() { - return adapterConfigs; - } - - public void setAdapterConfigs(List adapterConfigs) { - this.adapterConfigs = adapterConfigs; - - if (adapterConfigs != null) { - AdapterConfigs.configs.clear(); - for (String adapterConfig : adapterConfigs) { - int idx = adapterConfig.indexOf("/"); - if (idx > -1) { - String type = adapterConfig.substring(0, idx); - String ymlFile = adapterConfig.substring(idx + 1); - AdapterConfigs.configs.put(type, ymlFile); - } - } - } - } -} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java new file mode 100644 index 00000000..b7944852 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java @@ -0,0 +1,36 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.springframework.stereotype.Component; + +@Component +public class CuratorClient { + + @Resource + private AdapterCanalConfig adapterCanalConfig; + + private CuratorFramework curator = null; + + @PostConstruct + public void init() { + if (adapterCanalConfig.getZookeeperHosts() != null) { + curator = CuratorFrameworkFactory.builder() + .connectString(adapterCanalConfig.getZookeeperHosts()) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .sessionTimeoutMs(6000) + .connectionTimeoutMs(3000) + .namespace("canal-adapter") + .build(); + curator.start(); + } + } + + public CuratorFramework getCurator() { + return curator; + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/SpringContext.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/SpringContext.java new file mode 100644 index 00000000..ac860bea --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/SpringContext.java @@ -0,0 +1,28 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.stereotype.Component; + +@Component +public class SpringContext implements ApplicationContextAware { + + private static ApplicationContext context; + + /* + * 注入ApplicationContext + */ + public void setApplicationContext(final ApplicationContext context) throws BeansException { + // 在加载Spring时自动获得context + SpringContext.context = context; + } + + public static Object getBean(final String beanName) { + return SpringContext.context.getBean(beanName); + } + + public static Object getBean(final Class clz) { + return context.getBean(clz); + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index e0912bab..bd3a47a4 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -7,10 +7,14 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; +import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; +import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; +import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.protocol.FlatMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.MessageUtil; import com.alibaba.otter.canal.protocol.Message; @@ -24,34 +28,72 @@ public abstract class AbstractCanalAdapterWorker { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); - protected String canalDestination; // canal实例 - protected List> canalOuterAdapters; // 外部适配器 - protected ExecutorService groupInnerExecutorService; // 组内工作线程池 - protected volatile boolean running = false; // 是否运行中 + protected String canalDestination; // canal实例 + protected List> canalOuterAdapters; // 外部适配器 + protected ExecutorService groupInnerExecutorService; // 组内工作线程池 + protected volatile boolean running = false; // 是否运行中 protected Thread thread = null; - protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { + protected Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error", e); - @Override - public void uncaughtException(Thread t, Throwable e) { - logger.error("parse events has an error", e); - } - }; + protected SyncSwitch syncSwitch; + + public AbstractCanalAdapterWorker(){ + syncSwitch = (SyncSwitch) SpringContext.getBean(SyncSwitch.class); + } protected void writeOut(final Message message) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; + futures.add(groupInnerExecutorService.submit(() -> { + try { + // 组内适配器穿行运行，尽量不要配置组内适配器 + for (final OuterAdapter c : adapters) { + long begin = System.currentTimeMillis(); + MessageUtil.parse4Dml(canalDestination, message, c::sync); + + if (logger.isDebugEnabled()) { + logger.debug("{} elapsed time: {}", + c.getClass().getName(), + (System.currentTimeMillis() - begin)); + } + } + return true; + } catch (Exception e) { + return false; + } + })); + + // 等待所有适配器写入完成 + // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 + for (Future f : futures) { + try { + if (!f.get()) { + logger.error("Outer adapter write failed"); + } + } catch (InterruptedException | ExecutionException e) { + // ignore + } + } + } + } + + protected void writeOut(final FlatMessage flatMessage) { + List> futures = new ArrayList<>(); + // 组间适配器并行运行 + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; futures.add(groupInnerExecutorService.submit(new Callable() { @Override public Boolean call() { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (final CanalOuterAdapter c : adapters) { + for (OuterAdapter c : adapters) { long begin = System.currentTimeMillis(); - MessageUtil.parse4Dml(message, c::writeOut); - + Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage); + c.sync(dml); if (logger.isDebugEnabled()) { logger.debug("{} elapsed time: {}", c.getClass().getName(), @@ -79,63 +121,6 @@ public abstract class AbstractCanalAdapterWorker { } } - // protected void writeOut(final FlatMessage flatMessage) { - // List> futures = new ArrayList<>(); - // // 组间适配器并行运行 - // for (List outerAdapters : canalOuterAdapters) { - // final List adapters = outerAdapters; - // futures.add(groupInnerExecutorService.submit(new Callable() { - // - // @Override - // public Boolean call() { - // try { - // // 组内适配器穿行运行，尽量不要配置组内适配器 - // for (CanalOuterAdapter c : adapters) { - // long begin = System.currentTimeMillis(); - // Dml dml = MessageUtil.flatMessage2Dml(flatMessage); - // c.writeOut(dml); - // if (logger.isDebugEnabled()) { - // logger.debug("{} elapsed time: {}", - // c.getClass().getName(), - // (System.currentTimeMillis() - begin)); - // } - // } - // return true; - // } catch (Exception e) { - // return false; - // } - // } - // })); - // - // // 等待所有适配器写入完成 - // // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 - // for (Future f : futures) { - // try { - // if (!f.get()) { - // logger.error("Outer adapter write failed"); - // } - // } catch (InterruptedException | ExecutionException e) { - // // ignore - // } - // } - // } - // } - - protected void writeOut(Message message, String topic) { - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size()); - } - long begin = System.currentTimeMillis(); - writeOut(message); - long now = System.currentTimeMillis(); - if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { - logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); - } - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); - } - } - protected void stopOutAdapters() { if (thread != null) { try { @@ -146,8 +131,8 @@ public abstract class AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java new file mode 100644 index 00000000..22272e80 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -0,0 +1,142 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.consumer.CommitFailedException; +import org.apache.kafka.common.errors.WakeupException; + +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; +import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.protocol.Message; + +/** + * kafka对应的client适配器工作线程 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { + + private KafkaCanalConnector connector; + + private String topic; + + private boolean flatMessage; + + public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, + List> canalOuterAdapters, boolean flatMessage){ + this.canalOuterAdapters = canalOuterAdapters; + this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + this.topic = topic; + this.canalDestination = topic; + this.flatMessage = flatMessage; + connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage); + // connector.setSessionTimeout(1L, TimeUnit.MINUTES); + + // super.initSwitcher(topic); + } + + @Override + public void start() { + if (!running) { + thread = new Thread(() -> process()); + thread.setUncaughtExceptionHandler(handler); + running = true; + thread.start(); + } + } + + @Override + public void stop() { + try { + if (!running) { + return; + } + + connector.stopRunning(); + running = false; + + // if (switcher != null && !switcher.state()) { + // switcher.set(true); + // } + + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + groupInnerExecutorService.shutdown(); + logger.info("topic {} connectors' worker thread dead!", this.topic); + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { + adapter.destroy(); + } + } + logger.info("topic {} all connectors destroyed!", this.topic); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void process() { + while (!running) + ; + ExecutorService executor = Executors.newSingleThreadExecutor(); + // final AtomicBoolean executing = new AtomicBoolean(true); + while (running) { + try { + logger.info("=============> Start to connect topic: {} <=============", this.topic); + connector.connect(); + logger.info("=============> Start to subscribe topic: {} <=============", this.topic); + connector.subscribe(); + logger.info("=============> Subscribe topic: {} succeed <=============", this.topic); + while (running) { + try { + // switcher.get(); //等待开关开启 + + List messages; + if (!flatMessage) { + messages = connector.getWithoutAck(); + } else { + messages = connector.getFlatMessageWithoutAck(100L, TimeUnit.MILLISECONDS); + } + if (messages != null) { + for (final Object message : messages) { + if (message instanceof FlatMessage) { + writeOut((FlatMessage) message); + } else { + writeOut((Message) message); + } + } + } + connector.ack(); + } catch (CommitFailedException e) { + logger.warn(e.getMessage()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + TimeUnit.SECONDS.sleep(1L); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + executor.shutdown(); + + try { + connector.unsubscribe(); + } catch (WakeupException e) { + // No-op. Continue process + } + connector.disconnect(); + logger.info("=============> Disconnect topic: {} <=============", this.topic); + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 71c9d130..3bb5b4a4 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -9,13 +9,14 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; -import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; /** * 外部适配器的加载器 @@ -24,15 +25,17 @@ import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; */ public class CanalAdapterLoader { - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); - private CanalClientConfig canalClientConfig; + private CanalClientConfig canalClientConfig; - private Map canalWorkers = new HashMap<>(); + private Map canalWorkers = new HashMap<>(); - private ExtensionLoader loader; + private Map canalMQWorker = new HashMap<>(); - public CanalAdapterLoader(CanalClientConfig canalClientConfig) { + private ExtensionLoader loader; + + public CanalAdapterLoader(CanalClientConfig canalClientConfig){ this.canalClientConfig = canalClientConfig; } @@ -40,12 +43,7 @@ public class CanalAdapterLoader { * 初始化canal-client */ public void init() { - // canal instances 和 mq topics 配置不能同时为空 - // if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) { - // throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); - // } - - loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class); + loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class); String canalServerHost = this.canalClientConfig.getCanalServerHost(); SocketAddress sa = null; @@ -58,31 +56,67 @@ public class CanalAdapterLoader { // 初始化canal-client的适配器 if (canalClientConfig.getCanalInstances() != null) { for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new ArrayList<>(); for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { - List canalOutConnectors = new ArrayList<>(); - for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) { + List canalOutConnectors = new ArrayList<>(); + for (OuterAdapterConfig c : connectorGroup.getOutAdapters()) { loadConnector(c, canalOutConnectors); } canalOuterAdapterGroups.add(canalOutConnectors); } CanalAdapterWorker worker; - if (zkHosts != null) { + if (sa != null) { + worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); + } else if (zkHosts != null) { worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups); } else { - worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); + throw new RuntimeException("No canal server connector found"); } canalWorkers.put(instance.getInstance(), worker); worker.start(); logger.info("Start adapter for canal instance: {} succeed", instance.getInstance()); } } + + // 初始化canal-client-mq的适配器 + if (canalClientConfig.getMqTopics() != null) { + for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { + for (CanalClientConfig.Group group : topic.getGroups()) { + List> canalOuterAdapterGroups = new ArrayList<>(); + + List canalOuterAdapters = new ArrayList<>(); + + for (OuterAdapterConfig config : group.getOutAdapters()) { + loadConnector(config, canalOuterAdapters); + } + canalOuterAdapterGroups.add(canalOuterAdapters); + if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) { + CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig + .getBootstrapServers(), topic.getTopic(), group.getGroupId(), canalOuterAdapterGroups); + canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); + rocketMQWorker.start(); + } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) { + CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker( + canalClientConfig.getBootstrapServers(), + topic.getTopic(), + group.getGroupId(), + canalOuterAdapterGroups, + canalClientConfig.getFlatMessage()); + canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker); + canalKafkaWorker.start(); + } + logger.info("Start adapter for canal-client rocketmq topic: {} succeed", + topic.getTopic() + "-" + group.getGroupId()); + + } + } + } } - private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { + private void loadConnector(OuterAdapterConfig config, List canalOutConnectors) { try { - CanalOuterAdapter adapter = loader.getExtension(config.getName()); + OuterAdapter adapter = loader.getExtension(config.getName()); ClassLoader cl = Thread.currentThread().getContextClassLoader(); // 替换ClassLoader Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); @@ -103,10 +137,19 @@ public class CanalAdapterLoader { ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); for (CanalAdapterWorker v : canalWorkers.values()) { final CanalAdapterWorker caw = v; - stopExecutorService.submit(() -> caw.stop()); + stopExecutorService.submit(caw::stop); } stopExecutorService.shutdown(); } + + if (canalMQWorker.size() > 0) { + ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size()); + for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) { + final AbstractCanalAdapterWorker worker = tmp; + stopMQWokerService.submit(worker::stop); + } + stopMQWokerService.shutdown(); + } logger.info("All canal adapters destroyed"); } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java new file mode 100644 index 00000000..09863a33 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -0,0 +1,132 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.consumer.CommitFailedException; +import org.apache.kafka.common.errors.WakeupException; + +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; +import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; +import com.alibaba.otter.canal.protocol.Message; + +/** + * kafka对应的client适配器工作线程 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { + + private RocketMQCanalConnector connector; + + private String topic; + + public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, + List> canalOuterAdapters){ + logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); + this.canalOuterAdapters = canalOuterAdapters; + this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + this.topic = topic; + this.canalDestination = topic; + connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId); + } + + @Override + public void start() { + if (!running) { + thread = new Thread(new Runnable() { + + @Override + public void run() { + process(); + } + }); + thread.setUncaughtExceptionHandler(handler); + running = true; + thread.start(); + } + } + + @Override + public void stop() { + try { + if (!running) { + return; + } + connector.stopRunning(); + running = false; + logger.info("Stop topic {} out adapters begin", this.topic); + stopOutAdapters(); + logger.info("Stop topic {} out adapters end", this.topic); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void process() { + while (!running) + ; + ExecutorService executor = Executors.newSingleThreadExecutor(); + while (running) { + try { + logger.info("=============> Start to connect topic: {} <=============", this.topic); + connector.connect(); + logger.info("=============> Start to subscribe topic: {}<=============", this.topic); + connector.subscribe(); + logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); + while (running) { + try { + // switcher.get(); //等待开关开启 + + final Message message = connector.getWithoutAck(1); + if (message != null) { + executor.submit(new Runnable() { + + @Override + public void run() { + try { + if (logger.isDebugEnabled()) { + logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size()); + } + long begin = System.currentTimeMillis(); + writeOut(message); + long now = System.currentTimeMillis(); + if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { + logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + connector.ack(message.getId()); + } + }); + } else { + logger.debug("Message is null"); + } + } catch (CommitFailedException e) { + logger.warn(e.getMessage()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + TimeUnit.SECONDS.sleep(1L); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + executor.shutdown(); + + try { + connector.unsubscribe(); + } catch (WakeupException e) { + // No-op. Continue process + } + connector.stopRunning(); + logger.info("=============> Disconnect topic: {} <=============", this.topic); + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java new file mode 100644 index 00000000..03f515bf --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -0,0 +1,71 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import javax.annotation.Resource; + +import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; +import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; +import com.alibaba.otter.canal.adapter.launcher.config.AdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; + +@Component +public class CanalAdapterService { + + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterService.class); + + private static CanalAdapterLoader adapterLoader; + + @Resource + private AdapterCanalConfig adapterCanalConfig; + + // 注入bean保证优先注册 + @Resource + private AdapterConfig adapterConfig; + @Resource + private SpringContext springContext; + @Resource + private SyncSwitch syncSwitch; + + @PostConstruct + public void init() { + if (adapterLoader == null) { + try { + logger.info("## start the canal client adapters."); + adapterLoader = new CanalAdapterLoader(adapterCanalConfig); + adapterLoader.init(); + logger.info("## the canal client adapters are running now ......"); + } catch (Throwable e) { + logger.error("## something goes wrong when starting up the canal client adapters:", e); + System.exit(0); + } + } + } + + @PreDestroy + public void destroy() { + try { + logger.info("## stop the canal client adapters"); + if (adapterLoader != null) { + adapterLoader.destroy(); + } + for (DruidDataSource druidDataSource : DatasourceConfig.DATA_SOURCES.values()) { + try { + druidDataSource.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } catch (Throwable e) { + logger.warn("## something goes wrong when stopping canal client adapters:", e); + } finally { + logger.info("## canal client adapters are down."); + } + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index 7e090c2d..1c324996 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -3,10 +3,12 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.net.SocketAddress; import java.util.List; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.protocol.Message; @@ -31,7 +33,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, SocketAddress address, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -46,26 +48,18 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, String zookeeperHosts, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", ""); ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT); - - // super.initSwitcher(canalDestination); } @Override public void start() { if (!running) { - thread = new Thread(new Runnable() { - - @Override - public void run() { - process(); - } - }); + thread = new Thread(this::process); thread.setUncaughtExceptionHandler(handler); thread.start(); running = true; @@ -79,13 +73,11 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { return; } - // if (switcher != null && !switcher.state()) { - // switcher.set(true); - // } - connector.stopRunning(); running = false; + syncSwitch.release(canalDestination); + logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination); if (thread != null) { try { @@ -96,8 +88,8 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("destination {} adapters' worker thread dead!", canalDestination); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } @@ -112,22 +104,22 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { ; // waiting until running == true while (running) { try { - // if (switcher != null) { - // switcher.get(); - // } + syncSwitch.get(canalDestination); + logger.info("=============> Start to connect destination: {} <=============", this.canalDestination); connector.connect(); logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination); connector.subscribe(); logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination); while (running) { - // try { - // if (switcher != null) { - // switcher.get(); - // } - // } catch (TimeoutException e) { - // break; - // } + try { + syncSwitch.get(canalDestination, 1L, TimeUnit.MINUTES); + } catch (TimeoutException e) { + break; + } + if (!running) { + break; + } // server配置canal.instance.network.soTimeout(默认: 30s) // 范围内未与server交互，server将关闭本次socket连接 diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java new file mode 100644 index 00000000..bee78e04 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java @@ -0,0 +1,166 @@ +package com.alibaba.otter.canal.adapter.launcher.rest; + +import java.util.*; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.web.bind.annotation.*; + +import com.alibaba.otter.canal.adapter.launcher.common.EtlLock; +import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; +import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; +import com.alibaba.otter.canal.client.adapter.support.Result; + +@RestController +public class CommonRest { + + private static Logger logger = LoggerFactory.getLogger(CommonRest.class); + + private static final String ETL_LOCK_ZK_NODE = "/sync-etl/"; + + private ExtensionLoader loader; + + @Resource + private SyncSwitch syncSwitch; + @Resource + private EtlLock etlLock; + + @Resource + private AdapterCanalConfig adapterCanalConfig; + + @PostConstruct + public void init() { + loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class); + } + + /** + * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST + * + * @param type 类型 hbase, es + * @param task 任务名对应配置文件名 mytest_person2.yml + * @param params etl where条件参数, 为空全部导入 + * @return + */ + @PostMapping("/etl/{type}/{task}") + public EtlResult etl(@PathVariable String type, @PathVariable String task, + @RequestParam(name = "params", required = false) String params) { + + boolean locked = etlLock.tryLock(ETL_LOCK_ZK_NODE + type + "-" + task); + if (!locked) { + EtlResult result = new EtlResult(); + result.setSucceeded(false); + result.setErrorMessage(task + " 有其他进程正在导入中, 请稍后再试"); + return result; + } + try { + OuterAdapter adapter = loader.getExtension(type); + String destination = adapter.getDestination(task); + Boolean oriSwithcStatus = null; + if (destination != null) { + oriSwithcStatus = syncSwitch.status(destination); + syncSwitch.off(destination); + } + try { + List paramArr = null; + if (params != null) { + String[] parmaArray = params.trim().split(";"); + paramArr = Arrays.asList(parmaArray); + } + return adapter.etl(task, paramArr); + } finally { + if (destination != null && oriSwithcStatus != null && oriSwithcStatus) { + syncSwitch.on(destination); + } + } + } finally { + etlLock.unlock(ETL_LOCK_ZK_NODE + type + "-" + task); + } + } + + /** + * 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml + * + * @param type 类型 hbase, es + * @param task 任务名对应配置文件名 mytest_person2.yml + * @return + */ + @GetMapping("/count/{type}/{task}") + public Map count(@PathVariable String type, @PathVariable String task) { + OuterAdapter adapter = loader.getExtension(type); + return adapter.count(task); + } + + /** + * 返回所有实例 curl http://127.0.0.1:8081/destinations + */ + @GetMapping("/destinations") + public List> destinations() { + List> result = new ArrayList<>(); + Set destinations = adapterCanalConfig.DESTINATIONS; + for (String destination : destinations) { + Map resMap = new LinkedHashMap<>(); + Boolean status = syncSwitch.status(destination); + String resStatus = "none"; + if (status != null && status) { + resStatus = "on"; + } else if (status != null && !status) { + resStatus = "off"; + } + resMap.put("destination", destination); + resMap.put("status", resStatus); + result.add(resMap); + } + return result; + } + + /** + * 实例同步开关 curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT + * + * @param destination 实例名称 + * @param status 开关状态: off on + * @return + */ + @PutMapping("/syncSwitch/{destination}/{status}") + public Result etl(@PathVariable String destination, @PathVariable String status) { + if (status.equals("on")) { + syncSwitch.on(destination); + logger.info("#Destination: {} sync on", destination); + return Result.createSuccess("实例: " + destination + " 开启同步成功"); + } else if (status.equals("off")) { + syncSwitch.off(destination); + logger.info("#Destination: {} sync off", destination); + return Result.createSuccess("实例: " + destination + " 关闭同步成功"); + } else { + Result result = new Result(); + result.setCode(50000); + result.setMessage("实例: " + destination + " 操作失败"); + return result; + } + } + + /** + * 获取实例开关状态 curl http://127.0.0.1:8081/syncSwitch/example + * + * @param destination 实例名称 + * @return + */ + @GetMapping("/syncSwitch/{destination}") + public Map etl(@PathVariable String destination) { + Boolean status = syncSwitch.status(destination); + String resStatus = "none"; + if (status != null && status) { + resStatus = "on"; + } else if (status != null && !status) { + resStatus = "off"; + } + Map res = new LinkedHashMap<>(); + res.put("stauts", resStatus); + return res; + } +} diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index bb06e75b..43cc4e38 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -1,21 +1,47 @@ server: port: 8081 +logging: + level: + com.alibaba.otter.canal.client.adapter.hbase: DEBUG +spring: + jackson: + date-format: yyyy-MM-dd HH:mm:ss + time-zone: GMT+8 + default-property-inclusion: non_null + + +hbase.zookeeper.quorum: slave1 +hbase.zookeeper.property.clientPort: 2181 +hbase.zookeeper.znode.parent: /hbase-unsecure canal.conf: canalServerHost: 127.0.0.1:11111 - #zookeeperHosts: slave1:2181 - #bootstrapServers: localhost:9092 #or rocketmq nameservers:host1:9876;host2:9876 - flatMessage: false - +# zookeeperHosts: slave1:2181 +# bootstrapServers: slave1:6667 #or rocketmq nameservers:host1:9876;host2:9876 + flatMessage: true canalInstances: - instance: example adapterGroups: - outAdapters: - name: logger # - name: hbase -# hosts: slave1:2181 -# properties: {znodeParent: "/hbase-unsecure"} +# properties: +# hbase.zookeeper.quorum: ${hbase.zookeeper.quorum} +# hbase.zookeeper.property.clientPort: ${hbase.zookeeper.property.clientPort} +# zookeeper.znode.parent: ${hbase.zookeeper.znode.parent} +# mqTopics: +# - mqMode: kafka +# topic: example +# groups: +# - groupId: g2 +# outAdapters: +# - name: logger -adapter.conf.path: - adapterConfigs: - - hbase/mytest_person2.yml \ No newline at end of file +#adapter.conf: +# datasourceConfigs: +# defaultDS: +# url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true +# username: root +# password: 121212 +# adapterConfigs: +# - hbase/mytest_person2.yml diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java index 1885d6c2..985c7804 100644 --- a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java +++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java @@ -3,8 +3,8 @@ package com.alibaba.otter.canal.client.adapter.logger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.SPI; @@ -16,17 +16,17 @@ import com.alibaba.otter.canal.client.adapter.support.SPI; */ @SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name -public class LoggerAdapterExample implements CanalOuterAdapter { +public class LoggerAdapterExample implements OuterAdapter { private Logger logger = LoggerFactory.getLogger(this.getClass()); @Override - public void init(CanalOuterAdapterConfiguration configuration) { + public void init(OuterAdapterConfig configuration) { } @Override - public void writeOut(Dml dml) { + public void sync(Dml dml) { logger.info(dml.toString()); } diff --git a/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter similarity index 100% rename from client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter rename to client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter