excludeColumns) {
+ this.excludeColumns = excludeColumns;
+ }
+
public String getFamily() {
return family;
}
@@ -325,10 +349,10 @@ public class MappingConfig {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- HbaseOrm hbaseOrm = (HbaseOrm) o;
+ HbaseMapping hbaseMapping = (HbaseMapping) o;
- if (table != null ? !table.equals(hbaseOrm.table) : hbaseOrm.table != null) return false;
- return hbaseTable != null ? hbaseTable.equals(hbaseOrm.hbaseTable) : hbaseOrm.hbaseTable == null;
+ if (table != null ? !table.equals(hbaseMapping.table) : hbaseMapping.table != null) return false;
+ return hbaseTable != null ? hbaseTable.equals(hbaseMapping.hbaseTable) : hbaseMapping.hbaseTable == null;
}
@Override
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 929e9d6a..c0e9496a 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
@@ -4,6 +4,7 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -12,21 +13,19 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+
/**
* HBase表映射配置加载器
- *
- * 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。
- * 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式
- *
*
- * @author machengyuan 2018-8-21 下午06:45:49
+ * @author rewerma 2018-8-21 下午06:45:49
* @version 1.0.0
*/
public class MappingConfigLoader {
private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class);
- private static final String BASE_PATH = "hbase-mapping/";
+ private static final String BASE_PATH = "hbase";
/**
* 加载HBase表映射配置
@@ -35,12 +34,11 @@ public class MappingConfigLoader {
*/
public static Map load() {
logger.info("## Start loading mapping config ... ");
- String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf");
Map result = new LinkedHashMap<>();
- String[] configLines = mappingConfigContent.split("\n");
- for (String c : configLines) {
+ Collection configs = AdapterConfigs.get("hbase");
+ for (String c : configs) {
if (c == null) {
continue;
}
@@ -74,6 +72,7 @@ public class MappingConfigLoader {
String[] dbTable;
if (dsKey == null) {
dbTable = srcMeta.split("\\.");
+
} else {
dbTable = srcMeta.split("@")[0].split("\\.");
}
@@ -81,21 +80,22 @@ public class MappingConfigLoader {
if (dbTable.length == 2) {
config = new MappingConfig();
- MappingConfig.HbaseOrm hbaseOrm = new MappingConfig.HbaseOrm();
- hbaseOrm.setHbaseTable(dbTable[0].toUpperCase() + "." + dbTable[1].toUpperCase());
- hbaseOrm.setAutoCreateTable(true);
- hbaseOrm.setDatabase(dbTable[0]);
- hbaseOrm.setTable(dbTable[1]);
- hbaseOrm.setMode(MappingConfig.Mode.STRING);
- hbaseOrm.setRowKey(rowKey);
+ MappingConfig.HbaseMapping hbaseMapping = new MappingConfig.HbaseMapping();
+ hbaseMapping.setHbaseTable(dbTable[0].toUpperCase() + "." + dbTable[1].toUpperCase());
+ hbaseMapping.setAutoCreateTable(true);
+ hbaseMapping.setDatabase(dbTable[0]);
+ hbaseMapping.setTable(dbTable[1]);
+ hbaseMapping.setMode(MappingConfig.Mode.PHOENIX);
+ hbaseMapping.setRowKey(rowKey);
// 有定义rowKey
if (rowKey != null) {
MappingConfig.ColumnItem columnItem = new MappingConfig.ColumnItem();
columnItem.setRowKey(true);
columnItem.setColumn(rowKey);
- hbaseOrm.setRowKeyColumn(columnItem);
+ hbaseMapping.setRowKeyColumn(columnItem);
}
- config.setHbaseOrm(hbaseOrm);
+ config.setHbaseMapping(hbaseMapping);
+ 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..7c57a9ab
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java
@@ -0,0 +1,385 @@
+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;
+
+/**
+ * HBase ETL 操作业务类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+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;
+ }
+ }
+
+ /**
+ * 建表
+ *
+ * @param hbaseTemplate
+ * @param config
+ */
+ public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) {
+ try {
+ // 判断hbase表是否存在，不存在则建表
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
+ if (!hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
+ hbaseTemplate.createTable(hbaseMapping.getHbaseTable(), hbaseMapping.getFamily());
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * 导入数据
+ *
+ * @param ds 数据源
+ * @param hbaseTemplate hbaseTemplate
+ * @param config 配置
+ * @param params 筛选条件
+ * @return 导入结果
+ */
+ 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.HbaseMapping hbaseMapping = config.getHbaseMapping();
+ hbaseTable = hbaseMapping.getHbaseTable();
+
+ long start = System.currentTimeMillis();
+
+ if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) {
+ logger.info(hbaseMapping.getHbaseTable() + " rebuild is starting!");
+ // 如果表存在则删除
+ if (hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
+ hbaseTemplate.disableTable(hbaseMapping.getHbaseTable());
+ hbaseTemplate.deleteTable(hbaseMapping.getHbaseTable());
+ }
+ params = null;
+ } else {
+ logger.info(hbaseMapping.getHbaseTable() + " etl is starting!");
+ }
+ createTable(hbaseTemplate, config);
+
+ // 拼接sql
+ String sql = "SELECT * FROM " + config.getHbaseMapping().getDatabase() + "." + hbaseMapping.getTable();
+
+ // 拼接条件
+ if (params != null && params.size() == 1 && hbaseMapping.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 (hbaseMapping.getEtlCondition() != null && params != null) {
+ String etlCondition = hbaseMapping.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, hbaseMapping, hbaseTemplate, successCount, errMsg));
+ futures.add(future);
+ }
+
+ for (Future future : futures) {
+ future.get();
+ }
+
+ executor.shutdown();
+ } else {
+ executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg);
+ }
+
+ logger.info(
+ hbaseMapping.getHbaseTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!");
+
+ etlResult.setResultMessage("导入HBase表 " + hbaseMapping.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;
+ }
+
+ /**
+ * 执行导入
+ *
+ * @param ds
+ * @param sql
+ * @param hbaseMapping
+ * @param hbaseTemplate
+ * @param successCount
+ * @param errMsg
+ * @return
+ */
+ private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping,
+ 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 (hbaseMapping.getRowKey() != null) {
+ rowKeyColumns = hbaseMapping.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 = hbaseMapping.getColumnItems().get(columnName);
+ // 没有配置映射
+ if (columnItem == null) {
+ String family = hbaseMapping.getFamily();
+ String qualifile = columnName;
+ if (hbaseMapping.isUppercaseQualifier()) {
+ qualifile = qualifile.toUpperCase();
+ }
+ if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
+ if (hbaseMapping.getRowKey() == null && j == 1) {
+ row.setRowKey(Bytes.toBytes(val.toString()));
+ } else {
+ row.addCell(family, qualifile, Bytes.toBytes(val.toString()));
+ }
+ } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
+ Type type = Type.getType(classes[j - 1]);
+ if (hbaseMapping.getRowKey() == null && j == 1) {
+ row.setRowKey(TypeUtil.toBytes(val, type));
+ } else {
+ row.addCell(family, qualifile, TypeUtil.toBytes(val, type));
+ }
+ } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
+ PhType phType = PhType.getType(classes[j - 1]);
+ if (hbaseMapping.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 % hbaseMapping.getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseMapping.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(hbaseMapping.getHbaseTable(), rows);
+ }
+
+ } catch (Exception e) {
+ logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e);
+ errMsg.add(hbaseMapping.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..3dd03b91 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,29 +1,19 @@
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;
/**
* HBase同步操作业务
*
- * @author machengyuan 2018-8-21 下午06:45:49
+ * @author rewerma 2018-8-21 下午06:45:49
* @version 1.0.0
*/
public class HbaseSyncService {
@@ -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) {
@@ -69,11 +59,11 @@ public class HbaseSyncService {
return;
}
- MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
// if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists",
- // hbaseOrm.getHbaseTable());
+ // hbaseMapping.getHbaseTable());
// return;
// }
int i = 1;
@@ -83,28 +73,28 @@ public class HbaseSyncService {
HRow hRow = new HRow();
// 拼接复合rowKey
- if (hbaseOrm.getRowKey() != null) {
- String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+ if (hbaseMapping.getRowKey() != null) {
+ String[] rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
String rowKeyVale = getRowKeys(rowKeyColumns, r);
// params.put("rowKey", Bytes.toBytes(rowKeyVale));
hRow.setRowKey(Bytes.toBytes(rowKeyVale));
}
- convertData2Row(hbaseOrm, hRow, r);
+ convertData2Row(hbaseMapping, hRow, r);
if (hRow.getRowKey() == null) {
throw new RuntimeException("empty rowKey");
}
rows.add(hRow);
complete = false;
- if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
- hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ if (i % config.getHbaseMapping().getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
rows.clear();
complete = true;
}
i++;
}
if (!complete && !rows.isEmpty()) {
- hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
}
}
@@ -112,27 +102,30 @@ public class HbaseSyncService {
/**
* 将Map数据转换为HRow行数据
*
- * @param hbaseOrm hbase映射配置
+ * @param hbaseMapping hbase映射配置
* @param hRow 行对象
* @param data Map数据
*/
- private static void convertData2Row(MappingConfig.HbaseOrm hbaseOrm, HRow hRow, Map data) {
- Map columnItems = hbaseOrm.getColumnItems();
+ private static void convertData2Row(MappingConfig.HbaseMapping hbaseMapping, HRow hRow, Map data) {
+ Map columnItems = hbaseMapping.getColumnItems();
int i = 0;
for (Map.Entry entry : data.entrySet()) {
+ if (hbaseMapping.getExcludeColumns() != null && hbaseMapping.getExcludeColumns().contains(entry.getKey())) {
+ continue;
+ }
if (entry.getValue() != null) {
MappingConfig.ColumnItem columnItem = columnItems.get(entry.getKey());
- byte[] bytes = typeConvert(columnItem, hbaseOrm, entry.getValue());
+ byte[] bytes = typeConvert(columnItem, hbaseMapping, entry.getValue());
if (columnItem == null) {
- String familyName = hbaseOrm.getFamily();
+ String familyName = hbaseMapping.getFamily();
String qualifier = entry.getKey();
- if (hbaseOrm.isUppercaseQualifier()) {
+ if (hbaseMapping.isUppercaseQualifier()) {
qualifier = qualifier.toUpperCase();
}
- if (hbaseOrm.getRowKey() == null && i == 0) {
+ if (hbaseMapping.getRowKey() == null && i == 0) {
hRow.setRowKey(bytes);
} else {
hRow.addCell(familyName, qualifier, bytes);
@@ -163,15 +156,15 @@ public class HbaseSyncService {
return;
}
- MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
// if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists",
- // hbaseOrm.getHbaseTable());
+ // hbaseMapping.getHbaseTable());
// return;
// }
- MappingConfig.ColumnItem rowKeyColumn = hbaseOrm.getRowKeyColumn();
+ MappingConfig.ColumnItem rowKeyColumn = hbaseMapping.getRowKeyColumn();
int index = 0;
int i = 1;
boolean complete = false;
@@ -179,8 +172,8 @@ public class HbaseSyncService {
out: for (Map r : data) {
byte[] rowKeyBytes;
- if (hbaseOrm.getRowKey() != null) {
- String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+ if (hbaseMapping.getRowKey() != null) {
+ String[] rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
// 判断是否有复合主键修改
for (String updateColumn : old.get(index).keySet()) {
@@ -197,20 +190,23 @@ public class HbaseSyncService {
rowKeyBytes = Bytes.toBytes(rowKeyVale);
} else if (rowKeyColumn == null) {
Map rowKey = data.get(0);
- rowKeyBytes = typeConvert(null, hbaseOrm, rowKey.values().iterator().next());
+ rowKeyBytes = typeConvert(null, hbaseMapping, rowKey.values().iterator().next());
} else {
- rowKeyBytes = typeConvert(rowKeyColumn, hbaseOrm, r.get(rowKeyColumn.getColumn()));
+ rowKeyBytes = typeConvert(rowKeyColumn, hbaseMapping, r.get(rowKeyColumn.getColumn()));
}
if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
- Map columnItems = hbaseOrm.getColumnItems();
+ Map columnItems = hbaseMapping.getColumnItems();
HRow hRow = new HRow(rowKeyBytes);
for (String updateColumn : old.get(index).keySet()) {
+ if (hbaseMapping.getExcludeColumns() != null && hbaseMapping.getExcludeColumns().contains(updateColumn)) {
+ continue;
+ }
MappingConfig.ColumnItem columnItem = columnItems.get(updateColumn);
if (columnItem == null) {
- String family = hbaseOrm.getFamily();
+ String family = hbaseMapping.getFamily();
String qualifier = updateColumn;
- if (hbaseOrm.isUppercaseQualifier()) {
+ if (hbaseMapping.isUppercaseQualifier()) {
qualifier = qualifier.toUpperCase();
}
@@ -219,7 +215,7 @@ public class HbaseSyncService {
if (newVal == null) {
hRow.addCell(family, qualifier, null);
} else {
- hRow.addCell(family, qualifier, typeConvert(null, hbaseOrm, newVal));
+ hRow.addCell(family, qualifier, typeConvert(null, hbaseMapping, newVal));
}
} else {
// 排除修改id的情况
@@ -231,14 +227,14 @@ public class HbaseSyncService {
} else {
hRow.addCell(columnItem.getFamily(),
columnItem.getQualifier(),
- typeConvert(columnItem, hbaseOrm, newVal));
+ typeConvert(columnItem, hbaseMapping, newVal));
}
}
}
rows.add(hRow);
complete = false;
- if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
- hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ if (i % config.getHbaseMapping().getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
rows.clear();
complete = true;
}
@@ -246,7 +242,7 @@ public class HbaseSyncService {
index++;
}
if (!complete && !rows.isEmpty()) {
- hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
}
}
@@ -256,45 +252,45 @@ public class HbaseSyncService {
return;
}
- MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
// if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists",
- // hbaseOrm.getHbaseTable());
+ // hbaseMapping.getHbaseTable());
// return;
// }
- MappingConfig.ColumnItem rowKeyColumn = hbaseOrm.getRowKeyColumn();
+ MappingConfig.ColumnItem rowKeyColumn = hbaseMapping.getRowKeyColumn();
boolean complete = false;
int i = 1;
Set rowKeys = new HashSet<>();
for (Map r : data) {
byte[] rowKeyBytes;
- if (hbaseOrm.getRowKey() != null) {
- String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+ if (hbaseMapping.getRowKey() != null) {
+ String[] rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
String rowKeyVale = getRowKeys(rowKeyColumns, r);
rowKeyBytes = Bytes.toBytes(rowKeyVale);
} else if (rowKeyColumn == null) {
// 如果不需要类型转换
Map rowKey = data.get(0);
- rowKeyBytes = typeConvert(null, hbaseOrm, rowKey.values().iterator().next());
+ rowKeyBytes = typeConvert(null, hbaseMapping, rowKey.values().iterator().next());
} else {
Object val = r.get(rowKeyColumn.getColumn());
- rowKeyBytes = typeConvert(rowKeyColumn, hbaseOrm, val);
+ rowKeyBytes = typeConvert(rowKeyColumn, hbaseMapping, val);
}
if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
rowKeys.add(rowKeyBytes);
complete = false;
- if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rowKeys.isEmpty()) {
- hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+ if (i % config.getHbaseMapping().getCommitBatch() == 0 && !rowKeys.isEmpty()) {
+ hbaseTemplate.deletes(hbaseMapping.getHbaseTable(), rowKeys);
rowKeys.clear();
complete = true;
}
i++;
}
if (!complete && !rowKeys.isEmpty()) {
- hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+ hbaseTemplate.deletes(hbaseMapping.getHbaseTable(), rowKeys);
}
}
@@ -304,9 +300,9 @@ public class HbaseSyncService {
if (old == null || old.isEmpty() || data == null || data.isEmpty()) {
return;
}
- MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
- String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+ String[] rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
int index = 0;
int i = 1;
@@ -347,13 +343,13 @@ public class HbaseSyncService {
rowKeys.add(oldRowKeyBytes);
HRow row = new HRow(newRowKeyBytes);
- convertData2Row(hbaseOrm, row, r);
+ convertData2Row(hbaseMapping, row, r);
rows.add(row);
complete = false;
- if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
- hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+ if (i % config.getHbaseMapping().getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.deletes(hbaseMapping.getHbaseTable(), rowKeys);
- hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
rowKeys.clear();
rows.clear();
complete = true;
@@ -362,8 +358,8 @@ public class HbaseSyncService {
index++;
}
if (!complete && !rows.isEmpty()) {
- hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
- hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ hbaseTemplate.deletes(hbaseMapping.getHbaseTable(), rowKeys);
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
}
}
@@ -371,31 +367,32 @@ public class HbaseSyncService {
* 根据对应的类型进行转换
*
* @param columnItem 列项配置
- * @param hbaseOrm hbase映射配置
+ * @param hbaseMapping hbase映射配置
* @param value 值
* @return 复合字段rowKey
*/
- private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm, Object value) {
+ private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseMapping hbaseMapping,
+ Object value) {
if (value == null) {
return null;
}
byte[] bytes = null;
if (columnItem == null || columnItem.getType() == null || "".equals(columnItem.getType())) {
- if (MappingConfig.Mode.STRING == hbaseOrm.getMode()) {
+ if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
bytes = Bytes.toBytes(value.toString());
- } else if (MappingConfig.Mode.NATIVE == hbaseOrm.getMode()) {
+ } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
bytes = TypeUtil.toBytes(value);
- } else if (MappingConfig.Mode.PHOENIX == hbaseOrm.getMode()) {
+ } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
PhType phType = PhType.getType(value.getClass());
bytes = PhTypeUtil.toBytes(value, phType);
}
} else {
- if (hbaseOrm.getMode() == MappingConfig.Mode.STRING) {
+ if (hbaseMapping.getMode() == MappingConfig.Mode.STRING) {
bytes = Bytes.toBytes(value.toString());
- } else if (hbaseOrm.getMode() == MappingConfig.Mode.NATIVE) {
+ } else if (hbaseMapping.getMode() == MappingConfig.Mode.NATIVE) {
Type type = Type.getType(columnItem.getType());
bytes = TypeUtil.toBytes(value, type);
- } else if (hbaseOrm.getMode() == MappingConfig.Mode.PHOENIX) {
+ } else if (hbaseMapping.getMode() == MappingConfig.Mode.PHOENIX) {
PhType phType = PhType.getType(columnItem.getType());
bytes = PhTypeUtil.toBytes(value, phType);
}
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-mapping/configs.conf b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf
deleted file mode 100644
index ff5b292e..00000000
--- a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf
+++ /dev/null
@@ -1,5 +0,0 @@
-# 详细映射配置
-mytest_person2.yml
-
-# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认
-mytest.person
\ No newline at end of file
diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml
similarity index 92%
rename from client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml
rename to client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml
index 575a1bf5..f31341a3 100644
--- a/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml
+++ b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml
@@ -1,5 +1,7 @@
-hbaseOrm:
+dataSourceKey: defaultDS
+hbaseMapping:
mode: PHOENIX #NATIVE #STRING
+ destination: example
database: mytest # 数据库名
table: person2 # 数据库表名
hbaseTable: MYTEST.PERSON2 # HBase表名
@@ -15,6 +17,8 @@ hbaseOrm:
type: $DECIMAL
c_time: C_TIME$UNSIGNED_TIMESTAMP
birthday: BIRTHDAY$DATE
+ excludeColumns:
+ - lat # 忽略字段
# -- NATIVE类型
# $DEFAULT
@@ -43,7 +47,7 @@ hbaseOrm:
# $TINYINT 对应PHOENIX里的TINYINT 1字节
# $SMALLINT 对应PHOENIX里的SMALLINT 2字节
# $FLOAT 对应PHOENIX里的FLOAT 4字节
-# DOUBLE 对应PHOENIX里的DOUBLE 8字节
+# $DOUBLE 对应PHOENIX里的DOUBLE 8字节
# $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节
# $TIME 对应PHOENIX里的TIME 8字节
# $DATE 对应PHOENIX里的DATE 8字节
diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml
new file mode 100644
index 00000000..8159bb66
--- /dev/null
+++ b/client-adapter/launcher/pom.xml
@@ -0,0 +1,209 @@
+
+
+
+ canal.client-adapter
+ com.alibaba.otter
+ 1.1.1-SNAPSHOT
+
+ 4.0.0
+ com.alibaba.otter
+ client-adapter.launcher
+ jar
+ canal client adapter launcher module for otter ${project.version}
+
+
+
+
+ org.springframework.boot
+ spring-boot-dependencies
+ 2.0.1.RELEASE
+ pom
+ import
+
+
+
+
+
+
+ com.alibaba.otter
+ client-adapter.common
+ ${project.version}
+
+
+ com.alibaba.otter
+ canal.client
+ ${canal_version}
+
+
+ org.yaml
+ snakeyaml
+ 1.19
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+ org.springframework.boot
+ 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
+
+
+
+
+ com.alibaba.otter
+ client-adapter.logger
+ ${project.version}
+
+
+ *
+ *
+
+
+ jar-with-dependencies
+ true
+
+
+ com.alibaba.otter
+ client-adapter.hbase
+ ${project.version}
+
+
+ *
+ *
+
+
+ jar-with-dependencies
+ true
+
+
+
+
+ canal-adapter-launcher
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+ 2.0.1.RELEASE
+
+
+
+ com.alibaba.otter
+ client-adapter.hbase
+
+
+ com.alibaba.otter
+ client-adapter.logger
+
+
+
+
+
+
+ repackage
+
+
+
+
+
+ maven-antrun-plugin
+
+
+ package
+
+ run
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+ 2.10
+
+
+ copy-dependencies-to-canal-client-service
+ package
+
+ copy-dependencies
+
+
+ jar-with-dependencies
+ ${project.basedir}/target/lib
+
+
+
+
+
+
+
+
+
+ release
+
+
+ env
+ release
+
+
+
+
+
+
+ maven-assembly-plugin
+
+ false
+
+ ${basedir}/src/main/assembly/release.xml
+
+ ${project.artifactId}-${project.version}
+ ${project.basedir}/../../target
+
+
+
+ make-assembly
+ package
+
+ single
+
+
+
+
+
+
+
+
+
diff --git a/client-adapter/launcher/src/main/assembly/release.xml b/client-adapter/launcher/src/main/assembly/release.xml
new file mode 100644
index 00000000..7bb12314
--- /dev/null
+++ b/client-adapter/launcher/src/main/assembly/release.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/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
new file mode 100644
index 00000000..99140b0c
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java
@@ -0,0 +1,18 @@
+package com.alibaba.otter.canal.adapter.launcher;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+
+/**
+ * 启动入口
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@SpringBootApplication
+public class CanalAdapterApplication {
+
+ public static void main(String[] args) {
+ new SpringApplicationBuilder(CanalAdapterApplication.class).run(args);
+ }
+}
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..5fd5f3fe
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java
@@ -0,0 +1,120 @@
+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;
+
+/**
+ * Etl 同步锁
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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..aa6e55bc
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java
@@ -0,0 +1,213 @@
+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;
+
+/**
+ * 同步开关
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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
new file mode 100644
index 00000000..2579e1f7
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java
@@ -0,0 +1,51 @@
+package com.alibaba.otter.canal.adapter.launcher.config;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+
+/**
+ * canal 的相关配置类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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..f3e265f8
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java
@@ -0,0 +1,86 @@
+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;
+
+/**
+ * 适配器数据源及配置文件列表配置类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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("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/CuratorClient.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java
new file mode 100644
index 00000000..0ec60392
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java
@@ -0,0 +1,42 @@
+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;
+
+/**
+ * curator 配置类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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..987371e2
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/SpringContext.java
@@ -0,0 +1,34 @@
+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;
+
+/**
+ * spring util配置类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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
new file mode 100644
index 00000000..0486045c
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
@@ -0,0 +1,159 @@
+package com.alibaba.otter.canal.adapter.launcher.loader;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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.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;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * 适配器工作线程抽象类
+ *
+ * @author rewerma 2018-8-19 下午11:30:49
+ * @version 1.0.0
+ */
+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 Thread thread = null;
+ protected Thread.UncaughtExceptionHandler handler = (t, 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<>();
+ // 组间适配器并行运行
+ canalOuterAdapters.forEach(outerAdapters -> {
+ final List adapters = outerAdapters;
+ futures.add(groupInnerExecutorService.submit(() -> {
+ try {
+ // 组内适配器穿行运行，尽量不要配置组内适配器
+ adapters.forEach(adapter -> {
+ long begin = System.currentTimeMillis();
+ MessageUtil.parse4Dml(canalDestination, message, adapter::sync);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} elapsed time: {}",
+ adapter.getClass().getName(),
+ (System.currentTimeMillis() - begin));
+ }
+ });
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }));
+
+ // 等待所有适配器写入完成
+ // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
+ futures.forEach(future -> {
+ try {
+ if (!future.get()) {
+ logger.error("Outer adapter write failed");
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ // ignore
+ }
+ });
+ });
+ }
+
+ protected void writeOut(final FlatMessage flatMessage) {
+ List> futures = new ArrayList<>();
+ // 组间适配器并行运行
+ canalOuterAdapters.forEach(outerAdapters -> {
+ futures.add(groupInnerExecutorService.submit(() -> {
+ try {
+ // 组内适配器穿行运行，尽量不要配置组内适配器
+ outerAdapters.forEach(adapter -> {
+ long begin = System.currentTimeMillis();
+ Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage);
+ adapter.sync(dml);
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} elapsed time: {}",
+ adapter.getClass().getName(),
+ (System.currentTimeMillis() - begin));
+ }
+ });
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }));
+
+ // 等待所有适配器写入完成
+ // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
+ futures.forEach(future -> {
+ try {
+ if (!future.get()) {
+ logger.error("Outer adapter write failed");
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ // ignore
+ }
+ });
+ });
+ }
+
+ public void start() {
+ if (!running) {
+ thread = new Thread(this::process);
+ thread.setUncaughtExceptionHandler(handler);
+ thread.start();
+ running = true;
+ }
+ }
+
+ protected abstract void process();
+
+ public void stop() {
+ try {
+ if (!running) {
+ return;
+ }
+
+ closeConnection();
+ running = false;
+
+ syncSwitch.release(canalDestination);
+
+ logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination);
+ if (thread != null) {
+ try {
+ thread.join();
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+ groupInnerExecutorService.shutdown();
+ logger.info("destination {} adapters worker thread dead!", canalDestination);
+ canalOuterAdapters.forEach(outerAdapters -> outerAdapters.forEach(OuterAdapter::destroy));
+ logger.info("destination {} all adapters destroyed!", canalDestination);
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ protected abstract void closeConnection();
+
+}
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..665ad38e
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java
@@ -0,0 +1,101 @@
+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 rewerma 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);
+ }
+
+ @Override
+ protected void closeConnection(){
+ connector.stopRunning();
+ }
+
+ @Override
+ protected 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 {
+ syncSwitch.get(canalDestination);
+
+ 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-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
similarity index 61%
rename from client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java
rename to client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
index 6dd29c80..73606998 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.client.adapter.loader;
+package com.alibaba.otter.canal.adapter.launcher.loader;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
@@ -13,13 +13,13 @@ 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.ExtensionLoader;
+import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
/**
- * MQ外部适配器的加载器
+ * 外部适配器的加载器
*
* @version 1.0.0
*/
@@ -33,29 +33,17 @@ public class CanalAdapterLoader {
private Map canalMQWorker = new HashMap<>();
- private ExtensionLoader loader;
+ private ExtensionLoader loader;
public CanalAdapterLoader(CanalClientConfig canalClientConfig){
this.canalClientConfig = canalClientConfig;
}
/**
- * 初始化canal-client、 canal-client-rocketmq的适配器
+ * 初始化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, "" /*
- * TODO
- * canalClientConfig
- * .
- * getClassloaderPolicy
- * (
- * )
- */);
+ loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class);
String canalServerHost = this.canalClientConfig.getCanalServerHost();
SocketAddress sa = null;
@@ -68,20 +56,22 @@ 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();
@@ -93,11 +83,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);
@@ -109,7 +99,8 @@ public class CanalAdapterLoader {
canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
rocketMQWorker.start();
} else if ("kafka".equalsIgnoreCase(topic.getMqMode())) {
- CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig.getBootstrapServers(),
+ CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(
+ canalClientConfig.getBootstrapServers(),
topic.getTopic(),
group.getGroupId(),
canalOuterAdapterGroups,
@@ -117,17 +108,17 @@ public class CanalAdapterLoader {
canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker);
canalKafkaWorker.start();
}
- logger.info("Start adapter for canal-client rocketmq topic: {} succeed", topic.getTopic() + "-"
- + group.getGroupId());
+ 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());
@@ -146,31 +137,18 @@ public class CanalAdapterLoader {
public void destroy() {
if (canalWorkers.size() > 0) {
ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size());
- for (CanalAdapterWorker v : canalWorkers.values()) {
- final CanalAdapterWorker caw = v;
- stopExecutorService.submit(new Runnable() {
-
- @Override
- public void run() {
- caw.stop();
- }
- });
+ for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) {
+ stopExecutorService.submit(canalAdapterWorker::stop);
}
stopExecutorService.shutdown();
}
- if (canalMQWorker.size() > 0) {
- ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size());
- for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) {
- final AbstractCanalAdapterWorker worker = tmp;
- stopMQWokerService.submit(new Runnable() {
- @Override
- public void run() {
- worker.stop();
- }
- });
+ if (canalMQWorker.size() > 0) {
+ ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size());
+ for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) {
+ stopMQWorkerService.submit(canalAdapterMQWorker::stop);
}
- stopMQWokerService.shutdown();
+ stopMQWorkerService.shutdown();
}
logger.info("All canal adapters destroyed");
}
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
similarity index 70%
rename from client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
rename to client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
index 3c3b4abf..5b41e926 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.client.adapter.loader;
+package com.alibaba.otter.canal.adapter.launcher.loader;
import com.alibaba.otter.canal.protocol.FlatMessage;
import java.util.List;
@@ -9,7 +9,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;
@@ -17,7 +17,7 @@ import com.alibaba.otter.canal.protocol.Message;
/**
* kafka对应的client适配器工作线程
*
- * @author machengyuan 2018-8-19 下午11:30:49
+ * @author rewerma 2018-8-19 下午11:30:49
* @version 1.0.0
*/
public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
@@ -40,38 +40,12 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
}
@Override
- public void start() {
- if (!running) {
- thread = new Thread(new Runnable() {
-
- @Override
- public void run() {
- process();
- }
- });
- thread.setUncaughtExceptionHandler(handler);
- running = true;
- thread.start();
- }
+ protected void closeConnection() {
+ connector.stopRunning();
}
@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() {
+ protected void process() {
while (!running)
;
ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -91,6 +65,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
message = connector.getFlatMessageWithoutAck();
}
if (message != null) {
+<<<<<<< HEAD:client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
final Object msg = message;
executor.submit(new Runnable() {
@Override
@@ -110,7 +85,29 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
+=======
+ executor.submit(() -> {
+ 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);
+>>>>>>> 2c6bd3ee1b23f30fdb2b1f748805284a4fed872c:client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
}
+ connector.ack(message.getId());
});
} else {
logger.debug("Message is null");
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..2e535535
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java
@@ -0,0 +1,77 @@
+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;
+
+/**
+ * 适配器启动业务类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
similarity index 63%
rename from client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java
rename to client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
index e341c6a3..33a70f89 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
@@ -1,19 +1,21 @@
-package com.alibaba.otter.canal.client.adapter.loader;
+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;
/**
* 原生canal-server对应的client适配器工作线程
*
- * @author machengyuan 2018-8-19 下午11:30:49
+ * @author rewrema 2018-8-19 下午11:30:49
* @version 1.0.0
*/
public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
@@ -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,88 +48,41 @@ 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.setUncaughtExceptionHandler(handler);
- thread.start();
- running = true;
- }
+ protected void closeConnection() {
+ connector.stopRunning();
}
@Override
- public void stop() {
- try {
- if (!running) {
- return;
- }
-
- // if (switcher != null && !switcher.state()) {
- // switcher.set(true);
- // }
-
- connector.stopRunning();
- running = false;
-
- logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination);
- if (thread != null) {
- try {
- thread.join();
- } catch (InterruptedException e) {
- // ignore
- }
- }
- groupInnerExecutorService.shutdown();
- logger.info("destination {} adapters' worker thread dead!", canalDestination);
- for (List outerAdapters : canalOuterAdapters) {
- for (CanalOuterAdapter adapter : outerAdapters) {
- adapter.destroy();
- }
- }
- logger.info("destination {} all adapters destroyed!", canalDestination);
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- private void process() {
+ protected void process() {
while (!running)
; // 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连接
@@ -137,11 +92,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- // ignore
- }
+ Thread.sleep(1000);
} else {
if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} batchSize: {} ",
@@ -151,18 +102,18 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
}
long begin = System.currentTimeMillis();
writeOut(message);
- long now = System.currentTimeMillis();
if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} elapsed time: {} ms",
this.canalDestination,
batchId,
- now - begin);
+ System.currentTimeMillis() - begin);
}
}
connector.ack(batchId); // 提交确认
} catch (Exception e) {
connector.rollback(batchId); // 处理失败, 回滚数据
- throw e;
+ logger.error("sync error!", e);
+ Thread.sleep(500);
}
}
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..d2c163fb
--- /dev/null
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java
@@ -0,0 +1,172 @@
+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;
+
+/**
+ * 适配器操作Rest
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+@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
new file mode 100644
index 00000000..19fa6182
--- /dev/null
+++ b/client-adapter/launcher/src/main/resources/application.yml
@@ -0,0 +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
+
+
+hbasezookeeper.quorum: 127.0.0.1
+hbase.zookeeper.property.clientPort: 2181
+hbase.zookeeper.znode.parent: /hbase
+
+canal.conf:
+ canalServerHost: 127.0.0.1:11111
+# zookeeperHosts: slave1:2181
+# bootstrapServers: slave1:6667 #or rocketmq nameservers:host1:9876;host2:9876
+ flatMessage: true
+ canalInstances:
+ - instance: example
+ adapterGroups:
+ - outAdapters:
+ - name: logger
+# - name: hbase
+# 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:
+# 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/pom.xml b/client-adapter/logger/pom.xml
index 8f2b4edb..af7ec379 100644
--- a/client-adapter/logger/pom.xml
+++ b/client-adapter/logger/pom.xml
@@ -21,6 +21,7 @@