chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2 from alibaba/master

Browse Source 
update
This commit is contained in:
康智冬
2019-03-20 16:21:38 +08:00
committed by GitHub
parent 078673c6a9 b488ca02d8
commit 03d46f150f
49 changed files with 1629 additions and 1257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
.travlis.yml Normal file
View File

@@ -0,0 +1,23 @@
language: java
sudo: false # faster builds
jdk:
- openjdk11
- oraclejdk11
- openjdk8
- oraclejdk8
cache:
directories:
- $HOME/.m2
install: true
script:
- travis_wait 30 ./mvnw clean install -DskipTests=true -Dcheckstyle.skip=true -Drat.skip=true -Dmaven.javadoc.skip=true
after_success:
- bash <(curl -s https://codecov.io/bash)
after_failure:
- if [ -f canal.log ]; then echo "------TAIL of canal.log------"; tail -n 1000 canal.log; echo "------END of canal.log------"; fi

4
README.md
View File

@@ -82,7 +82,7 @@ See the wiki page for : <a href="https://github.com/alibaba/canal/wiki" >wiki文
1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR 1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR
* canal java客户端: <a href="https://github.com/alibaba/canal/wiki/ClientExample"> https://github.com/alibaba/canal/wiki/ClientExample </a> * canal java客户端: <a href="https://github.com/alibaba/canal/wiki/ClientExample"> https://github.com/alibaba/canal/wiki/ClientExample </a>
* canal c#客户端开源项目地址：<a href="https://github.com/CanalClient/CanalSharp"> https://github.com/CanalSharp/CanalSharp </a> * canal c#客户端开源项目地址：<a href="https://github.com/dotnetcore/CanalSharp"> https://github.com/dotnetcore/CanalSharp </a>
* canal go客户端开源项目地址：<a href="https://github.com/CanalClient/canal-go"> https://github.com/CanalClient/canal-go </a> * canal go客户端开源项目地址：<a href="https://github.com/CanalClient/canal-go"> https://github.com/CanalClient/canal-go </a>
2. canal作为MySQL binlog的增量获取工具，可以将数据投递到MQ系统中，比如Kafka/RocketMQ，可以借助于MQ的多语言能力 2. canal作为MySQL binlog的增量获取工具，可以将数据投递到MQ系统中，比如Kafka/RocketMQ，可以借助于MQ的多语言能力
* 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart) * 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart)
@@ -117,7 +117,7 @@ See the wiki page for : <a href="https://github.com/alibaba/canal/wiki" >wiki文
<h3>最新更新</h3> <h3>最新更新</h3>
<ol> <ol>
<li>canal发布重大版本更新1.1.0，具体releaseNode参考：<a href="https://github.com/alibaba/canal/releases/tag/canal-1.1.0">https://github.com/alibaba/canal/releases/tag/canal-1.1.0</a></li> <li>canal发布重大版本更新1.1.0，具体releaseNode参考：<a href="https://github.com/alibaba/canal/releases/tag/canal-1.1.0">https://github.com/alibaba/canal/releases/tag/canal-1.1.0</a></li>
<li>canal c#客户端开源项目地址：<a href="https://github.com/CanalClient/CanalSharp"> https://github.com/CanalClient/CanalSharp </a>，推荐! </li> <li>canal c#客户端开源项目地址：<a href="https://github.com/dotnetcore/CanalSharp"> https://github.com/dotnetcore/CanalSharp </a>，推荐! </li>
<li>canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。</li> <li>canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。</li>
<li>canal消费端项目开源: Otter(分布式数据库同步系统)，地址：<a href="https://github.com/alibaba/otter">https://github.com/alibaba/otter</a></li> <li>canal消费端项目开源: Otter(分布式数据库同步系统)，地址：<a href="https://github.com/alibaba/otter">https://github.com/alibaba/otter</a></li>

2
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
View File

@@ -84,7 +84,7 @@ public class YmlConfigBinder {
try { try {
byte[] contentBytes; byte[] contentBytes;
if (charset == null) { if (charset == null) {
contentBytes = content.getBytes(); contentBytes = content.getBytes("UTF-8");
} else { } else {
contentBytes = content.getBytes(charset); contentBytes = content.getBytes(charset);
} }

2
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/RelaxedConversionService.java
View File

@@ -89,7 +89,7 @@ class RelaxedConversionService implements ConversionService {
return new RelaxedConversionService.StringToEnumIgnoringCaseConverterFactory.StringToEnum(enumType); return new RelaxedConversionService.StringToEnumIgnoringCaseConverterFactory.StringToEnum(enumType);
} }
private class StringToEnum<T extends Enum> implements Converter<String, T> { private static class StringToEnum<T extends Enum> implements Converter<String, T> {
private final Class<T> enumType; private final Class<T> enumType;

1
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java
View File

@@ -191,6 +191,7 @@ public abstract class YamlProcessor {
} }
private boolean process(Map<String, Object> map, MatchCallback callback) { private boolean process(Map<String, Object> map, MatchCallback callback) {
@SuppressWarnings("serial")
Properties properties = new Properties() { Properties properties = new Properties() {
@Override @Override

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

@@ -6,6 +6,7 @@ import java.io.InputStreamReader;
import java.net.MalformedURLException; import java.net.MalformedURLException;
import java.net.URL; import java.net.URL;
import java.net.URLClassLoader; import java.net.URLClassLoader;
import java.nio.file.Path;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.util.Arrays; import java.util.Arrays;
import java.util.Enumeration; import java.util.Enumeration;
@@ -27,7 +28,8 @@ import org.slf4j.LoggerFactory;
*/ */
public class ExtensionLoader<T> { public class ExtensionLoader<T> {
private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class); private static final Logger logger = LoggerFactory
.getLogger(ExtensionLoader.class);
private static final String SERVICES_DIRECTORY = "META-INF/services/"; private static final String SERVICES_DIRECTORY = "META-INF/services/";
@@ -35,7 +37,8 @@ public class ExtensionLoader<T> {
private static final String DEFAULT_CLASSLOADER_POLICY = "internal"; private static final String DEFAULT_CLASSLOADER_POLICY = "internal";
private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*"); private static final Pattern NAME_SEPARATOR = Pattern
.compile("\\s*[,]+\\s*");
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>(); private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
@@ -171,7 +174,8 @@ public class ExtensionLoader<T> {
return instance; return instance;
} catch (Throwable t) { } catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+ ") could not be instantiated: " + t.getMessage(), t); + ") could not be instantiated: " + t.getMessage(),
t);
} }
} }
@@ -191,7 +195,8 @@ public class ExtensionLoader<T> {
return instance; return instance;
} catch (Throwable t) { } catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+ ") could not be instantiated: " + t.getMessage(), t); + ") could not be instantiated: " + t.getMessage(),
t);
} }
} }
@@ -233,7 +238,11 @@ public class ExtensionLoader<T> {
// button. // button.
jarPath = jarPath.replaceAll("/classes/.*", "/classes/"); jarPath = jarPath.replaceAll("/classes/.*", "/classes/");
} }
return Paths.get(jarPath).getParent().toString(); // Paths - from java 8 Path path = Paths.get(jarPath).getParent(); // Paths - from java 8
if (path != null) {
return path.toString();
}
return null;
} }
private Map<String, Class<?>> loadExtensionClasses() { private Map<String, Class<?>> loadExtensionClasses() {
@@ -330,12 +339,10 @@ public class ExtensionLoader<T> {
// Class.forName(line, true, // Class.forName(line, true,
// classLoader); // classLoader);
if (!type.isAssignableFrom(clazz)) { if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException("Error when load extension class(interface: " throw new IllegalStateException(
+ type "Error when load extension class(interface: " + type
+ ", class line: " + ", class line: " + clazz.getName()
+ clazz.getName() + "), class " + clazz.getName()
+ "), class "
+ clazz.getName()
+ "is not subtype of interface."); + "is not subtype of interface.");
} else { } else {
try { try {
@@ -353,9 +360,9 @@ public class ExtensionLoader<T> {
extensionClasses.put(n, clazz); extensionClasses.put(n, clazz);
} else if (c != clazz) { } else if (c != clazz) {
cachedNames.remove(clazz); cachedNames.remove(clazz);
throw new IllegalStateException("Duplicate extension " throw new IllegalStateException(
+ type.getName() "Duplicate extension " + type.getName() + " name "
+ " name " + n + " on " + n + " on "
+ c.getName() + " and " + c.getName() + " and "
+ clazz.getName()); + clazz.getName());
} }
@@ -365,12 +372,9 @@ public class ExtensionLoader<T> {
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: " IllegalStateException e = new IllegalStateException(
+ type "Failed to load extension class(interface: " + type + ", class line: "
+ ", class line: " + line + ") in " + url
+ line
+ ") in "
+ url
+ ", cause: " + ", cause: "
+ t.getMessage(), + t.getMessage(),
t); t);
@@ -385,13 +389,15 @@ public class ExtensionLoader<T> {
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", class file: " + url logger.error("Exception when load extension class(interface: " + type + ", class file: " + url
+ ") in " + url, t); + ") in " + url,
t);
} }
} // end of while urls } // end of while urls
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName logger.error(
+ ").", t); "Exception when load extension class(interface: " + type + ", description file: " + fileName + ").",
t);
} }
} }

2
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java
View File

@@ -155,9 +155,11 @@ public class JdbcTypeUtil {
} else { } else {
res = null; res = null;
} }
break;
case Types.CLOB: case Types.CLOB:
default: default:
res = value; res = value;
break;
} }
return res; return res;
} catch (Exception e) { } catch (Exception e) {

8
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java
View File

@@ -12,10 +12,10 @@ import java.util.Date;
public class Result implements Serializable { public class Result implements Serializable {
private static final long serialVersionUID = -3276409502352405716L; private static final long serialVersionUID = -3276409502352405716L;
public Integer code = 20000; private Integer code = 20000;
public Object data; private Object data;
public String message; private String message;
public Date sysTime; private Date sysTime;
public static Result createSuccess(String message) { public static Result createSuccess(String message) {
Result result = new Result(); Result result = new Result();

105
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java
View File

@@ -2,11 +2,17 @@ package com.alibaba.otter.canal.client.adapter.support;
import java.io.File; import java.io.File;
import java.net.URL; import java.net.URL;
import java.sql.*; import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.*; import java.time.*;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.Date; import java.util.Date;
import java.util.TimeZone; import java.util.TimeZone;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Function; import java.util.function.Function;
@@ -29,38 +35,16 @@ public class Util {
/** /**
* 通过DS执行sql * 通过DS执行sql
*/ */
public static Object sqlRS(DataSource ds, String sql, Function<ResultSet, Object> fun) throws SQLException { public static Object sqlRS(DataSource ds, String sql, Function<ResultSet, Object> fun) {
Connection conn = null; try (Connection conn = ds.getConnection();
Statement stmt = null; Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);) {
ResultSet rs = null;
try {
conn = ds.getConnection();
stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
stmt.setFetchSize(Integer.MIN_VALUE); stmt.setFetchSize(Integer.MIN_VALUE);
rs = stmt.executeQuery(sql); try (ResultSet rs = stmt.executeQuery(sql);) {
return fun.apply(rs); 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);
}
} }
} catch (Exception e) {
logger.error("sqlRs has error, sql: {} ", sql);
throw new RuntimeException(e);
} }
} }
@@ -72,29 +56,10 @@ public class Util {
* @param consumer 回调方法 * @param consumer 回调方法
*/ */
public static void sqlRS(Connection conn, String sql, Consumer<ResultSet> consumer) { public static void sqlRS(Connection conn, String sql, Consumer<ResultSet> consumer) {
Statement stmt = null; try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) {
ResultSet rs = null;
try {
stmt = conn.createStatement();
rs = stmt.executeQuery(sql);
consumer.accept(rs); consumer.accept(rs);
} catch (SQLException e) { } catch (SQLException e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
} 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);
}
}
} }
} }
@@ -144,7 +109,41 @@ public class Util {
return column; return column;
} }
private static String timeZone; // 当前时区 public static ThreadPoolExecutor newFixedThreadPool(int nThreads, long keepAliveTime) {
return new ThreadPoolExecutor(nThreads,
nThreads,
keepAliveTime,
TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
(r, exe) -> {
if (!exe.isShutdown()) {
try {
exe.getQueue().put(r);
} catch (InterruptedException e) {
// ignore
}
}
});
}
public static ThreadPoolExecutor newSingleThreadExecutor(long keepAliveTime) {
return new ThreadPoolExecutor(1,
1,
keepAliveTime,
TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
(r, exe) -> {
if (!exe.isShutdown()) {
try {
exe.getQueue().put(r);
} catch (InterruptedException e) {
// ignore
}
}
});
}
public final static String timeZone; // 当前时区
private static DateTimeZone dateTimeZone; private static DateTimeZone dateTimeZone;
static { static {
@@ -265,7 +264,7 @@ public class Util {
LocalDateTime localDateTime = LocalDateTime.of(localDate, localTime); LocalDateTime localDateTime = LocalDateTime.of(localDate, localTime);
return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant()); return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
} }
} catch (Exception e) { } catch (Throwable e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
} }

6
client-adapter/elasticsearch/pom.xml
View File

@@ -40,6 +40,12 @@
<version>4.12</version> <version>4.12</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.40</version>
<scope>test</scope>
</dependency>
</dependencies> </dependencies>
<build> <build>

4
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
View File

@@ -133,8 +133,8 @@ public class ESAdapter implements OuterAdapter {
esSyncService = new ESSyncService(esTemplate); esSyncService = new ESSyncService(esTemplate);
esConfigMonitor = new ESConfigMonitor(); esConfigMonitor = new ESConfigMonitor();
esConfigMonitor.init(this); esConfigMonitor.init(this, envProperties);
} catch (Exception e) { } catch (Throwable e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
} }

60
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
View File

@@ -31,7 +31,7 @@ public class ESSyncConfig {
throw new NullPointerException("esMapping._type"); throw new NullPointerException("esMapping._type");
} }
if (esMapping._id == null && esMapping.getPk() == null) { if (esMapping._id == null && esMapping.getPk() == null) {
throw new NullPointerException("esMapping._id and esMapping.pk"); throw new NullPointerException("esMapping._id or esMapping.pk");
} }
if (esMapping.sql == null) { if (esMapping.sql == null) {
throw new NullPointerException("esMapping.sql"); throw new NullPointerException("esMapping.sql");
@@ -80,23 +80,23 @@ public class ESSyncConfig {
public static class ESMapping { public static class ESMapping {
private String _index; private String _index;
private String _type; private String _type;
private String _id; private String _id;
private boolean upsert = false; private boolean upsert = false;
private String pk; private String pk;
// private String parent; private Map<String, RelationMapping> relations = new LinkedHashMap<>();
private String sql; private String sql;
// 对象字段, 例: objFields: // 对象字段, 例: objFields:
// - _labels: array:; // - _labels: array:;
private Map<String, String> objFields = new LinkedHashMap<>(); private Map<String, String> objFields = new LinkedHashMap<>();
private List<String> skips = new ArrayList<>(); private List<String> skips = new ArrayList<>();
private int commitBatch = 1000; private int commitBatch = 1000;
private String etlCondition; private String etlCondition;
private boolean syncByTimestamp = false; // 是否按时间戳定时同步 private boolean syncByTimestamp = false; // 是否按时间戳定时同步
private Long syncInterval; // 同步时间间隔 private Long syncInterval; // 同步时间间隔
private SchemaItem schemaItem; // sql解析结果模型 private SchemaItem schemaItem; // sql解析结果模型
public String get_index() { public String get_index() {
return _index; return _index;
@@ -154,6 +154,14 @@ public class ESSyncConfig {
this.skips = skips; this.skips = skips;
} }
public Map<String, RelationMapping> getRelations() {
return relations;
}
public void setRelations(Map<String, RelationMapping> relations) {
this.relations = relations;
}
public String getSql() { public String getSql() {
return sql; return sql;
} }
@@ -202,4 +210,26 @@ public class ESSyncConfig {
this.schemaItem = schemaItem; this.schemaItem = schemaItem;
} }
} }
public static class RelationMapping {
private String name;
private String parent;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getParent() {
return parent;
}
public void setParent(String parent) {
this.parent = parent;
}
}
} }

5
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
View File

@@ -278,12 +278,13 @@ public class SchemaItem {
if (relationSelectFieldItems == null) { if (relationSelectFieldItems == null) {
synchronized (SchemaItem.class) { synchronized (SchemaItem.class) {
if (relationSelectFieldItems == null) { if (relationSelectFieldItems == null) {
relationSelectFieldItems = new ArrayList<>(); List<FieldItem> relationSelectFieldItemsTmp = new ArrayList<>();
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
if (fieldItem.getOwners().contains(getAlias())) { if (fieldItem.getOwners().contains(getAlias())) {
relationSelectFieldItems.add(fieldItem); relationSelectFieldItemsTmp.add(fieldItem);
} }
} }
relationSelectFieldItems = relationSelectFieldItemsTmp;
} }
} }
} }

26
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java
View File

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.es.monitor;
import java.io.File; import java.io.File;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.Properties;
import java.util.regex.Matcher; import java.util.regex.Matcher;
import java.util.regex.Pattern; import java.util.regex.Pattern;
@@ -12,9 +13,9 @@ import org.apache.commons.io.monitor.FileAlterationMonitor;
import org.apache.commons.io.monitor.FileAlterationObserver; import org.apache.commons.io.monitor.FileAlterationObserver;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.es.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
@@ -31,10 +32,13 @@ public class ESConfigMonitor {
private ESAdapter esAdapter; private ESAdapter esAdapter;
private Properties envProperties;
private FileAlterationMonitor fileMonitor; private FileAlterationMonitor fileMonitor;
public void init(ESAdapter esAdapter) { public void init(ESAdapter esAdapter, Properties envProperties) {
this.esAdapter = esAdapter; this.esAdapter = esAdapter;
this.envProperties = envProperties;
File confDir = Util.getConfDirPath(adapterName); File confDir = Util.getConfDirPath(adapterName);
try { try {
FileAlterationObserver observer = new FileAlterationObserver(confDir, FileAlterationObserver observer = new FileAlterationObserver(confDir,
@@ -65,11 +69,13 @@ public class ESConfigMonitor {
try { try {
// 加载新增的配置文件 // 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
ESSyncConfig config = new Yaml().loadAs(configContent, ESSyncConfig.class); ESSyncConfig config = YmlConfigBinder
config.validate(); .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
addConfigToCache(file, config); if (config != null) {
config.validate();
logger.info("Add a new es mapping config: {} to canal adapter", file.getName()); addConfigToCache(file, config);
logger.info("Add a new es mapping config: {} to canal adapter", file.getName());
}
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
} }
@@ -88,7 +94,11 @@ public class ESConfigMonitor {
onFileDelete(file); onFileDelete(file);
return; return;
} }
ESSyncConfig config = new Yaml().loadAs(configContent, ESSyncConfig.class); ESSyncConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate(); config.validate();
if (esAdapter.getEsSyncConfig().containsKey(file.getName())) { if (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
deleteConfigFromCache(file); deleteConfigFromCache(file);

98
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
View File

@@ -1,25 +1,23 @@
package com.alibaba.otter.canal.client.adapter.es.service; package com.alibaba.otter.canal.client.adapter.es.service;
import java.util.ArrayList; import java.sql.SQLException;
import java.util.LinkedHashMap; import java.util.*;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher; import java.util.regex.Matcher;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import javax.sql.DataSource; import javax.sql.DataSource;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType; import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHit;
@@ -30,10 +28,10 @@ import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
import com.alibaba.otter.canal.client.adapter.es.support.ESSyncUtil;
import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.google.common.base.Joiner; import com.google.common.base.Joiner;
/** /**
@@ -100,7 +98,7 @@ public class ESEtlService {
// 获取总数 // 获取总数
String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> { long cnt = (Long) Util.sqlRS(dataSource, countSql, rs -> {
Long count = null; Long count = null;
try { try {
if (rs.next()) { if (rs.next()) {
@@ -116,8 +114,7 @@ public class ESEtlService {
if (cnt >= 10000) { if (cnt >= 10000) {
int threadCount = 3; // 从配置读取默认为3 int threadCount = 3; // 从配置读取默认为3
long perThreadCnt = cnt / threadCount; long perThreadCnt = cnt / threadCount;
ExecutorService executor = Executors.newFixedThreadPool(threadCount); ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
List<Future<Boolean>> futures = new ArrayList<>(threadCount);
for (int i = 0; i < threadCount; i++) { for (int i = 0; i < threadCount; i++) {
long offset = i * perThreadCnt; long offset = i * perThreadCnt;
Long size = null; Long size = null;
@@ -130,16 +127,13 @@ public class ESEtlService {
} else { } else {
sqlFinal = sql + " LIMIT " + offset + "," + cnt; sqlFinal = sql + " LIMIT " + offset + "," + cnt;
} }
Future<Boolean> future = executor executor.execute(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
.submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
futures.add(future);
}
for (Future<Boolean> future : futures) {
future.get();
} }
executor.shutdown(); executor.shutdown();
while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
// ignore
}
} else { } else {
executeSqlImport(dataSource, sql, mapping, impCount, errMsg); executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
} }
@@ -176,7 +170,7 @@ public class ESEtlService {
private boolean executeSqlImport(DataSource ds, String sql, ESMapping mapping, AtomicLong impCount, private boolean executeSqlImport(DataSource ds, String sql, ESMapping mapping, AtomicLong impCount,
List<String> errMsg) { List<String> errMsg) {
try { try {
ESSyncUtil.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
int count = 0; int count = 0;
try { try {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
@@ -184,39 +178,73 @@ public class ESEtlService {
long batchBegin = System.currentTimeMillis(); long batchBegin = System.currentTimeMillis();
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
Object idVal = null;
for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) { for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) {
// 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) {
continue;
}
String fieldName = fieldItem.getFieldName(); String fieldName = fieldItem.getFieldName();
if (mapping.getSkips().contains(fieldName)) { if (mapping.getSkips().contains(fieldName)) {
continue; continue;
} }
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); // 如果是主键字段则不插入
esFieldData.put(fieldName, val); if (fieldItem.getFieldName().equals(mapping.get_id())) {
idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
} else {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
esFieldData.put(Util.cleanColumn(fieldName), val);
}
} }
Object idVal = null;
if (mapping.get_id() != null) { if (!mapping.getRelations().isEmpty()) {
idVal = rs.getObject(mapping.get_id()); mapping.getRelations().forEach((relationField, relationMapping) -> {
Map<String, Object> relations = new HashMap<>();
relations.put("name", relationMapping.getName());
if (StringUtils.isNotEmpty(relationMapping.getParent())) {
FieldItem parentFieldItem = mapping.getSchemaItem()
.getSelectFields()
.get(relationMapping.getParent());
Object parentVal;
try {
parentVal = esTemplate.getValFromRS(mapping,
rs,
parentFieldItem.getFieldName(),
parentFieldItem.getFieldName());
} catch (SQLException e) {
throw new RuntimeException(e);
}
if (parentVal != null) {
relations.put("parent", parentVal.toString());
esFieldData.put("$parent_routing", parentVal.toString());
}
}
esFieldData.put(Util.cleanColumn(relationField), relations);
});
} }
if (idVal != null) { if (idVal != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
bulkRequestBuilder.add(transportClient UpdateRequestBuilder updateRequestBuilder = transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString()) .prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString())
.setDoc(esFieldData) .setDoc(esFieldData)
.setDocAsUpsert(true)); .setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
updateRequestBuilder.setRouting(parentVal);
}
bulkRequestBuilder.add(updateRequestBuilder);
} else { } else {
bulkRequestBuilder.add(transportClient IndexRequestBuilder indexRequestBuilder = transportClient
.prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString()) .prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString())
.setSource(esFieldData)); .setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
indexRequestBuilder.setRouting(parentVal);
}
bulkRequestBuilder.add(indexRequestBuilder);
} }
} else { } else {
idVal = rs.getObject(mapping.getPk()); idVal = esFieldData.get(mapping.getPk());
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) .setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))

33
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
View File

@@ -7,6 +7,7 @@ import java.util.Map;
import javax.sql.DataSource; import javax.sql.DataSource;
import com.alibaba.otter.canal.client.adapter.support.Util;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -43,7 +44,7 @@ public class ESSyncService {
long begin = System.currentTimeMillis(); long begin = System.currentTimeMillis();
if (esSyncConfigs != null) { if (esSyncConfigs != null) {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Destination: {}, database:{}, table:{}, type:{}, effect index count: {}", logger.trace("Destination: {}, database:{}, table:{}, type:{}, affected index count: {}",
dml.getDestination(), dml.getDestination(),
dml.getDatabase(), dml.getDatabase(),
dml.getTable(), dml.getTable(),
@@ -65,7 +66,7 @@ public class ESSyncService {
} }
} }
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Sync elapsed time: {} ms, effect index count：{}, destination: {}", logger.trace("Sync elapsed time: {} ms, affected indexes count：{}, destination: {}",
(System.currentTimeMillis() - begin), (System.currentTimeMillis() - begin),
esSyncConfigs.size(), esSyncConfigs.size(),
dml.getDestination()); dml.getDestination());
@@ -74,7 +75,7 @@ public class ESSyncService {
StringBuilder configIndexes = new StringBuilder(); StringBuilder configIndexes = new StringBuilder();
esSyncConfigs esSyncConfigs
.forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" ")); .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
logger.debug("DML: {} \nEffect indexes: {}", logger.debug("DML: {} \nAffected indexes: {}",
JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue), JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
configIndexes.toString()); configIndexes.toString());
} }
@@ -166,7 +167,7 @@ public class ESSyncService {
data, data,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
@@ -295,7 +296,7 @@ public class ESSyncService {
data, data,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
} }
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
@@ -407,7 +408,7 @@ public class ESSyncService {
// ------关联表简单字段更新为null------ // ------关联表简单字段更新为null------
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) { for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
esFieldData.put(fieldItem.getFieldName(), null); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null);
} }
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData); joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
} else { } else {
@@ -464,7 +465,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
sql.replace("\n", " ")); sql.replace("\n", " "));
} }
ESSyncUtil.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
try { try {
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
@@ -500,7 +501,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
sql.replace("\n", " ")); sql.replace("\n", " "));
} }
ESSyncUtil.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
try { try {
Map<String, Object> esFieldData = null; Map<String, Object> esFieldData = null;
if (mapping.getPk() != null) { if (mapping.getPk() != null) {
@@ -508,7 +509,7 @@ public class ESSyncService {
esTemplate.getESDataFromDmlData(mapping, data, esFieldData); esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
esFieldData.remove(mapping.getPk()); esFieldData.remove(mapping.getPk());
for (String key : esFieldData.keySet()) { for (String key : esFieldData.keySet()) {
esFieldData.put(key, null); esFieldData.put(Util.cleanColumn(key), null);
} }
} }
while (rs.next()) { while (rs.next()) {
@@ -601,7 +602,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
sql.toString().replace("\n", " ")); sql.toString().replace("\n", " "));
} }
ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { Util.sqlRS(ds, sql.toString(), rs -> {
try { try {
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
@@ -617,7 +618,7 @@ public class ESSyncService {
rs, rs,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
break out; break out;
} }
} }
@@ -628,7 +629,7 @@ public class ESSyncService {
rs, rs,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName()); fieldItem.getColumn().getColumnName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
} }
} }
@@ -693,7 +694,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
sql.toString().replace("\n", " ")); sql.toString().replace("\n", " "));
} }
ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { Util.sqlRS(ds, sql.toString(), rs -> {
try { try {
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
@@ -724,7 +725,7 @@ public class ESSyncService {
rs, rs,
fieldItem.getFieldName(), fieldItem.getFieldName(),
fieldItem.getFieldName()); fieldItem.getFieldName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
break; break;
} }
} }
@@ -733,7 +734,7 @@ public class ESSyncService {
} else { } else {
Object val = esTemplate Object val = esTemplate
.getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName()); .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
esFieldData.put(fieldItem.getFieldName(), val); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
} }
} }
@@ -812,7 +813,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
sql.replace("\n", " ")); sql.replace("\n", " "));
} }
ESSyncUtil.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
try { try {
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();

72
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java
View File

@@ -3,12 +3,9 @@ package com.alibaba.otter.canal.client.adapter.es.support;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.sql.*; import java.sql.Blob;
import java.sql.SQLException;
import java.util.*; import java.util.*;
import java.util.Date;
import java.util.function.Function;
import javax.sql.DataSource;
import org.apache.commons.codec.binary.Base64; import org.apache.commons.codec.binary.Base64;
import org.joda.time.DateTime; import org.joda.time.DateTime;
@@ -113,9 +110,9 @@ public class ESSyncUtil {
} else if (val instanceof java.sql.Timestamp) { } else if (val instanceof java.sql.Timestamp) {
DateTime dateTime = new DateTime(((java.sql.Timestamp) val).getTime()); DateTime dateTime = new DateTime(((java.sql.Timestamp) val).getTime());
if (dateTime.getMillisOfSecond() != 0) { if (dateTime.getMillisOfSecond() != 0) {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else { } else {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
} }
} else if (val instanceof java.sql.Date || val instanceof Date) { } else if (val instanceof java.sql.Date || val instanceof Date) {
DateTime dateTime; DateTime dateTime;
@@ -129,9 +126,9 @@ public class ESSyncUtil {
res = dateTime.toString("yyyy-MM-dd"); res = dateTime.toString("yyyy-MM-dd");
} else { } else {
if (dateTime.getMillisOfSecond() != 0) { if (dateTime.getMillisOfSecond() != 0) {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else { } else {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
} }
} }
} else if (val instanceof Long) { } else if (val instanceof Long) {
@@ -140,9 +137,9 @@ public class ESSyncUtil {
&& dateTime.getMillisOfSecond() == 0) { && dateTime.getMillisOfSecond() == 0) {
res = dateTime.toString("yyyy-MM-dd"); res = dateTime.toString("yyyy-MM-dd");
} else if (dateTime.getMillisOfSecond() != 0) { } else if (dateTime.getMillisOfSecond() != 0) {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else { } else {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
} }
} else if (val instanceof String) { } else if (val instanceof String) {
String v = ((String) val).trim(); String v = ((String) val).trim();
@@ -153,9 +150,9 @@ public class ESSyncUtil {
if (date != null) { if (date != null) {
DateTime dateTime = new DateTime(date); DateTime dateTime = new DateTime(date);
if (dateTime.getMillisOfSecond() != 0) { if (dateTime.getMillisOfSecond() != 0) {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else { } else {
res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00"); res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
} }
} }
} else if (v.length() == 10 && v.charAt(4) == '-' && v.charAt(7) == '-') { } else if (v.length() == 10 && v.charAt(4) == '-' && v.charAt(7) == '-') {
@@ -234,8 +231,11 @@ public class ESSyncUtil {
private static byte[] blobToBytes(Blob blob) { private static byte[] blobToBytes(Blob blob) {
try (InputStream is = blob.getBinaryStream()) { try (InputStream is = blob.getBinaryStream()) {
byte[] b = new byte[(int) blob.length()]; byte[] b = new byte[(int) blob.length()];
is.read(b); if (is.read(b) != -1) {
return b; return b;
} else {
return new byte[0];
}
} catch (IOException | SQLException e) { } catch (IOException | SQLException e) {
logger.error(e.getMessage()); logger.error(e.getMessage());
return null; return null;
@@ -297,46 +297,4 @@ public class ESSyncUtil {
sql.append(owner).append(".").append(columnName).append("=").append(value).append(" AND "); sql.append(owner).append(".").append(columnName).append("=").append(value).append(" AND ");
} }
} }
/**
* 执行查询sql
*/
public static Object sqlRS(DataSource ds, String sql, Function<ResultSet, Object> fun) {
Connection conn = null;
Statement smt = null;
ResultSet rs = null;
try {
conn = ds.getConnection();
smt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
smt.setFetchSize(Integer.MIN_VALUE);
rs = smt.executeQuery(sql);
return fun.apply(rs);
} catch (SQLException e) {
logger.error("sqlRs has error, sql: {} ", sql);
throw new RuntimeException(e);
} finally {
if (rs != null) {
try {
rs.close();
} catch (SQLException e) {
logger.error("error to close result set");
}
}
if (smt != null) {
try {
smt.close();
} catch (SQLException e) {
logger.error("error to close statement");
}
}
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
logger.error("error to close db connection");
}
}
}
}
} }

126
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
View File

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.client.adapter.es.support;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.HashMap;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
@@ -9,10 +10,13 @@ import java.util.concurrent.ConcurrentMap;
import javax.sql.DataSource; import javax.sql.DataSource;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.common.collect.ImmutableOpenMap;
@@ -29,6 +33,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
/** /**
* ES 操作模板 * ES 操作模板
@@ -64,13 +69,24 @@ public class ESTemplate {
*/ */
public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) UpdateRequestBuilder updateRequestBuilder = transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData) .setDoc(esFieldData)
.setDocAsUpsert(true)); .setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
updateRequestBuilder.setRouting(parentVal);
}
getBulk().add(updateRequestBuilder);
} else { } else {
getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) IndexRequestBuilder indexRequestBuilder = transportClient
.setSource(esFieldData)); .prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
indexRequestBuilder.setRouting(parentVal);
}
getBulk().add(indexRequestBuilder);
} }
commitBulk(); commitBulk();
} else { } else {
@@ -96,7 +112,9 @@ public class ESTemplate {
* @param esFieldData 数据Map * @param esFieldData 数据Map
*/ */
public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
append4Update(mapping, pkVal, esFieldData); Map<String, Object> esFieldDataTmp = new LinkedHashMap<>(esFieldData.size());
esFieldData.forEach((k, v) -> esFieldDataTmp.put(Util.cleanColumn(k), v));
append4Update(mapping, pkVal, esFieldDataTmp);
commitBulk(); commitBulk();
} }
@@ -122,7 +140,7 @@ public class ESTemplate {
(fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND ")); (fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND "));
int len = sql.length(); int len = sql.length();
sql.delete(len - 4, len); sql.delete(len - 4, len);
Integer syncCount = (Integer) ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), rs -> {
int count = 0; int count = 0;
try { try {
while (rs.next()) { while (rs.next()) {
@@ -137,7 +155,7 @@ public class ESTemplate {
return count; return count;
}); });
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Update ES by query effect {} records", syncCount); logger.trace("Update ES by query affected {} records", syncCount);
} }
} }
@@ -200,13 +218,24 @@ public class ESTemplate {
private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) UpdateRequestBuilder updateRequestBuilder = transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData) .setDoc(esFieldData)
.setDocAsUpsert(true)); .setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
updateRequestBuilder.setRouting(parentVal);
}
getBulk().add(updateRequestBuilder);
} else { } else {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) UpdateRequestBuilder updateRequestBuilder = transportClient
.setDoc(esFieldData)); .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
updateRequestBuilder.setRouting(parentVal);
}
getBulk().add(updateRequestBuilder);
} }
} else { } else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
@@ -223,6 +252,8 @@ public class ESTemplate {
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
String columnName) throws SQLException { String columnName) throws SQLException {
fieldName = Util.cleanColumn(fieldName);
columnName = Util.cleanColumn(columnName);
String esType = getEsType(mapping, fieldName); String esType = getEsType(mapping, fieldName);
Object value = resultSet.getObject(columnName); Object value = resultSet.getObject(columnName);
@@ -254,9 +285,13 @@ public class ESTemplate {
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
} }
// 添加父子文档关联信息
putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
return resultIdVal; return resultIdVal;
} }
@@ -288,12 +323,16 @@ public class ESTemplate {
for (ColumnItem columnItem : fieldItem.getColumnItems()) { for (ColumnItem columnItem : fieldItem.getColumnItems()) {
if (dmlOld.containsKey(columnItem.getColumnName()) if (dmlOld.containsKey(columnItem.getColumnName())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
break; break;
} }
} }
} }
// 添加父子文档关联信息
putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
return resultIdVal; return resultIdVal;
} }
@@ -337,9 +376,12 @@ public class ESTemplate {
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
} }
// 添加父子文档关联信息
putRelationData(mapping, schemaItem, dmlData, esFieldData);
return resultIdVal; return resultIdVal;
} }
@@ -364,13 +406,67 @@ public class ESTemplate {
} }
if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
} }
} }
// 添加父子文档关联信息
putRelationData(mapping, schemaItem, dmlOld, esFieldData);
return resultIdVal; return resultIdVal;
} }
private void putRelationDataFromRS(ESMapping mapping, SchemaItem schemaItem, ResultSet resultSet,
Map<String, Object> esFieldData) {
// 添加父子文档关联信息
if (!mapping.getRelations().isEmpty()) {
mapping.getRelations().forEach((relationField, relationMapping) -> {
Map<String, Object> relations = new HashMap<>();
relations.put("name", relationMapping.getName());
if (StringUtils.isNotEmpty(relationMapping.getParent())) {
FieldItem parentFieldItem = schemaItem.getSelectFields().get(relationMapping.getParent());
Object parentVal;
try {
parentVal = getValFromRS(mapping,
resultSet,
parentFieldItem.getFieldName(),
parentFieldItem.getFieldName());
} catch (SQLException e) {
throw new RuntimeException(e);
}
if (parentVal != null) {
relations.put("parent", parentVal.toString());
esFieldData.put("$parent_routing", parentVal.toString());
}
}
esFieldData.put(relationField, relations);
});
}
}
private void putRelationData(ESMapping mapping, SchemaItem schemaItem, Map<String, Object> dmlData,
Map<String, Object> esFieldData) {
// 添加父子文档关联信息
if (!mapping.getRelations().isEmpty()) {
mapping.getRelations().forEach((relationField, relationMapping) -> {
Map<String, Object> relations = new HashMap<>();
relations.put("name", relationMapping.getName());
if (StringUtils.isNotEmpty(relationMapping.getParent())) {
FieldItem parentFieldItem = schemaItem.getSelectFields().get(relationMapping.getParent());
String columnName = parentFieldItem.getColumnItems().iterator().next().getColumnName();
Object parentVal = getValFromData(mapping, dmlData, parentFieldItem.getFieldName(), columnName);
if (parentVal != null) {
relations.put("parent", parentVal.toString());
esFieldData.put("$parent_routing", parentVal.toString());
}
}
esFieldData.put(relationField, relations);
});
}
}
/** /**
* es 字段类型本地缓存 * es 字段类型本地缓存
*/ */

21
client-adapter/elasticsearch/src/main/resources/es/biz_order.yml Normal file
View File

@@ -0,0 +1,21 @@
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: customer
_type: _doc
_id: _id
relations:
customer_order:
name: order
parent: customer_id
sql: "select concat('oid_', t.id) as _id,
t.customer_id,
t.id as order_id,
t.serial_code as order_serial,
t.c_time as order_time
from biz_order t"
skips:
- customer_id
etlCondition: "where t.c_time>='{0}'"
commitBatch: 3000

47
client-adapter/elasticsearch/src/main/resources/es/customer.yml Normal file
View File

@@ -0,0 +1,47 @@
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: customer
_type: _doc
_id: id
relations:
customer_order:
name: customer
sql: "select t.id, t.name, t.email from customer t"
etlCondition: "where t.c_time>='{0}'"
commitBatch: 3000
#{
# "mappings":{
# "_doc":{
# "properties":{
# "id": {
# "type": "long"
# },
# "name": {
# "type": "text"
# },
# "email": {
# "type": "text"
# },
# "order_id": {
# "type": "long"
# },
# "order_serial": {
# "type": "text"
# },
# "order_time": {
# "type": "date"
# },
# "customer_order":{
# "type":"join",
# "relations":{
# "customer":"order"
# }
# }
# }
# }
# }
#}

1
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java
View File

@@ -23,6 +23,7 @@ public class ConfigLoadTest {
public void testLoad() { public void testLoad() {
Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load(null); Map<String, ESSyncConfig> configMap = ESSyncConfigLoader.load(null);
ESSyncConfig config = configMap.get("mytest_user.yml"); ESSyncConfig config = configMap.get("mytest_user.yml");
config.validate();
Assert.assertNotNull(config); Assert.assertNotNull(config);
Assert.assertEquals("defaultDS", config.getDataSourceKey()); Assert.assertEquals("defaultDS", config.getDataSourceKey());
ESSyncConfig.ESMapping esMapping = config.getEsMapping(); ESSyncConfig.ESMapping esMapping = config.getEsMapping();

118
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ESTest.java Normal file
View File

@@ -0,0 +1,118 @@
package com.alibaba.otter.canal.client.adapter.es.test;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.LinkedHashMap;
import java.util.Map;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
public class ESTest {
private TransportClient transportClient;
@Before
public void init() throws UnknownHostException {
Settings.Builder settingBuilder = Settings.builder();
settingBuilder.put("cluster.name", TestConstant.clusterName);
Settings settings = settingBuilder.build();
transportClient = new PreBuiltTransportClient(settings);
String[] hostArray = TestConstant.esHosts.split(",");
for (String host : hostArray) {
int i = host.indexOf(":");
transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)),
Integer.parseInt(host.substring(i + 1))));
}
}
@Test
public void test01() {
SearchResponse response = transportClient.prepareSearch("test")
.setTypes("osm")
.setQuery(QueryBuilders.termQuery("_id", "1"))
.setSize(10000)
.get();
for (SearchHit hit : response.getHits()) {
System.out.println(hit.getSourceAsMap().get("data").getClass());
}
}
@Test
public void test02() {
Map<String, Object> esFieldData = new LinkedHashMap<>();
esFieldData.put("userId", 2L);
esFieldData.put("eventId", 4L);
esFieldData.put("eventName", "网络异常");
esFieldData.put("description", "第四个事件信息");
Map<String, Object> relations = new LinkedHashMap<>();
esFieldData.put("user_event", relations);
relations.put("name", "event");
relations.put("parent", "2");
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
bulkRequestBuilder
.add(transportClient.prepareIndex("test", "osm", "2_4").setRouting("2").setSource(esFieldData));
commit(bulkRequestBuilder);
}
@Test
public void test03() {
Map<String, Object> esFieldData = new LinkedHashMap<>();
esFieldData.put("userId", 2L);
esFieldData.put("eventName", "网络异常1");
Map<String, Object> relations = new LinkedHashMap<>();
esFieldData.put("user_event", relations);
relations.put("name", "event");
relations.put("parent", "2");
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
bulkRequestBuilder.add(transportClient.prepareUpdate("test", "osm", "2_4").setRouting("2").setDoc(esFieldData));
commit(bulkRequestBuilder);
}
@Test
public void test04() {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
bulkRequestBuilder.add(transportClient.prepareDelete("test", "osm", "2_4"));
commit(bulkRequestBuilder);
}
private void commit(BulkRequestBuilder bulkRequestBuilder) {
if (bulkRequestBuilder.numberOfActions() > 0) {
BulkResponse response = bulkRequestBuilder.execute().actionGet();
if (response.hasFailures()) {
for (BulkItemResponse itemResponse : response.getItems()) {
if (!itemResponse.isFailed()) {
continue;
}
if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
System.out.println(itemResponse.getFailureMessage());
} else {
System.out.println("ES bulk commit error" + itemResponse.getFailureMessage());
}
}
}
}
}
@After
public void after() {
transportClient.close();
}
}

4
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/TestConstant.java
View File

@@ -11,9 +11,9 @@ public class TestConstant {
public final static String jdbcPassword = "121212"; public final static String jdbcPassword = "121212";
public final static String esHosts = "127.0.0.1:9300"; public final static String esHosts = "127.0.0.1:9300";
public final static String clusterNmae = "elasticsearch"; public final static String clusterName = "elasticsearch";
public static DruidDataSource dataSource; public final static DruidDataSource dataSource;
static { static {
dataSource = new DruidDataSource(); dataSource = new DruidDataSource();

2
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java
View File

@@ -22,7 +22,7 @@ public class Common {
outerAdapterConfig.setName("es"); outerAdapterConfig.setName("es");
outerAdapterConfig.setHosts(TestConstant.esHosts); outerAdapterConfig.setHosts(TestConstant.esHosts);
Map<String, String> properties = new HashMap<>(); Map<String, String> properties = new HashMap<>();
properties.put("cluster.name", TestConstant.clusterNmae); properties.put("cluster.name", TestConstant.clusterName);
outerAdapterConfig.setProperties(properties); outerAdapterConfig.setProperties(properties);
ESAdapter esAdapter = new ESAdapter(); ESAdapter esAdapter = new ESAdapter();

2
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
View File

@@ -96,7 +96,7 @@ public class HbaseAdapter implements OuterAdapter {
hbaseSyncService = new HbaseSyncService(hbaseTemplate); hbaseSyncService = new HbaseSyncService(hbaseTemplate);
configMonitor = new HbaseConfigMonitor(); configMonitor = new HbaseConfigMonitor();
configMonitor.init(this); configMonitor.init(this, envProperties);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }

20
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java
View File

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.hbase.monitor;
import java.io.File; import java.io.File;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.Properties;
import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
@@ -11,8 +12,8 @@ import org.apache.commons.io.monitor.FileAlterationObserver;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter; import com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader; import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
@@ -26,10 +27,13 @@ public class HbaseConfigMonitor {
private HbaseAdapter hbaseAdapter; private HbaseAdapter hbaseAdapter;
private Properties envProperties;
private FileAlterationMonitor fileMonitor; private FileAlterationMonitor fileMonitor;
public void init(HbaseAdapter hbaseAdapter) { public void init(HbaseAdapter hbaseAdapter, Properties envProperties) {
this.hbaseAdapter = hbaseAdapter; this.hbaseAdapter = hbaseAdapter;
this.envProperties = envProperties;
File confDir = Util.getConfDirPath(adapterName); File confDir = Util.getConfDirPath(adapterName);
try { try {
FileAlterationObserver observer = new FileAlterationObserver(confDir, FileAlterationObserver observer = new FileAlterationObserver(confDir,
@@ -60,7 +64,11 @@ public class HbaseConfigMonitor {
try { try {
// 加载新增的配置文件 // 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate(); config.validate();
addConfigToCache(file, config); addConfigToCache(file, config);
@@ -83,7 +91,11 @@ public class HbaseConfigMonitor {
onFileDelete(file); onFileDelete(file);
return; return;
} }
MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate(); config.validate();
if (hbaseAdapter.getHbaseMapping().containsKey(file.getName())) { if (hbaseAdapter.getHbaseMapping().containsKey(file.getName())) {
deleteConfigFromCache(file); deleteConfigFromCache(file);

788
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java
View File

@@ -1,397 +1,391 @@
package com.alibaba.otter.canal.client.adapter.hbase.service; package com.alibaba.otter.canal.client.adapter.hbase.service;
import java.sql.ResultSetMetaData; import java.sql.ResultSetMetaData;
import java.sql.Timestamp; import java.sql.Timestamp;
import java.text.SimpleDateFormat; import java.text.SimpleDateFormat;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
import java.util.concurrent.ExecutorService; import java.util.concurrent.*;
import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong; import javax.sql.DataSource;
import javax.sql.DataSource; import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.apache.hadoop.hbase.util.Bytes; import org.slf4j.LoggerFactory;
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.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
import com.alibaba.otter.canal.client.adapter.hbase.support.HRow; import com.alibaba.otter.canal.client.adapter.hbase.support.PhType;
import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate; import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil;
import com.alibaba.otter.canal.client.adapter.hbase.support.PhType; import com.alibaba.otter.canal.client.adapter.hbase.support.Type;
import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil; import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil;
import com.alibaba.otter.canal.client.adapter.hbase.support.Type; import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil; import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil;
import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil; import com.google.common.base.Joiner;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.google.common.base.Joiner; /**
* HBase ETL 操作业务类
/** *
* HBase ETL 操作业务类 * @author rewerma @ 2018-10-20
* * @version 1.0.0
* @author rewerma @ 2018-10-20 */
* @version 1.0.0 public class HbaseEtlService {
*/
public class HbaseEtlService { private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class);
private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class); /**
* 建表
*
/** * @param hbaseTemplate
* 建表 * @param config
* */
* @param hbaseTemplate public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) {
* @param config try {
*/ // 判断hbase表是否存在，不存在则建表
public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) { MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
try { if (!hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
// 判断hbase表是否存在，不存在则建表 hbaseTemplate.createTable(hbaseMapping.getHbaseTable(), hbaseMapping.getFamily());
MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping(); }
if (!hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) { } catch (Exception e) {
hbaseTemplate.createTable(hbaseMapping.getHbaseTable(), hbaseMapping.getFamily()); logger.error(e.getMessage(), e);
} throw new RuntimeException(e);
} catch (Exception e) { }
logger.error(e.getMessage(), e); }
throw new RuntimeException(e);
} /**
} * 导入数据
*
/** * @param ds 数据源
* 导入数据 * @param hbaseTemplate hbaseTemplate
* * @param config 配置
* @param ds 数据源 * @param params 筛选条件
* @param hbaseTemplate hbaseTemplate * @return 导入结果
* @param config 配置 */
* @param params 筛选条件 public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config,
* @return 导入结果 List<String> params) {
*/ EtlResult etlResult = new EtlResult();
public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config, AtomicLong successCount = new AtomicLong();
List<String> params) { List<String> errMsg = new ArrayList<>();
EtlResult etlResult = new EtlResult(); String hbaseTable = "";
AtomicLong successCount = new AtomicLong(); try {
List<String> errMsg = new ArrayList<>(); if (config == null) {
String hbaseTable = ""; logger.error("Config is null!");
try { etlResult.setSucceeded(false);
if (config == null) { etlResult.setErrorMessage("Config is null!");
logger.error("Config is null!"); return etlResult;
etlResult.setSucceeded(false); }
etlResult.setErrorMessage("Config is null!"); MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
return etlResult; hbaseTable = hbaseMapping.getHbaseTable();
}
MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping(); long start = System.currentTimeMillis();
hbaseTable = hbaseMapping.getHbaseTable();
if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) {
long start = System.currentTimeMillis(); logger.info(hbaseMapping.getHbaseTable() + " rebuild is starting!");
// 如果表存在则删除
if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) { if (hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
logger.info(hbaseMapping.getHbaseTable() + " rebuild is starting!"); hbaseTemplate.disableTable(hbaseMapping.getHbaseTable());
// 如果表存在则删除 hbaseTemplate.deleteTable(hbaseMapping.getHbaseTable());
if (hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) { }
hbaseTemplate.disableTable(hbaseMapping.getHbaseTable()); params = null;
hbaseTemplate.deleteTable(hbaseMapping.getHbaseTable()); } else {
} logger.info(hbaseMapping.getHbaseTable() + " etl is starting!");
params = null; }
} else { createTable(hbaseTemplate, config);
logger.info(hbaseMapping.getHbaseTable() + " etl is starting!");
} // 拼接sql
createTable(hbaseTemplate, config); String sql = "SELECT * FROM " + config.getHbaseMapping().getDatabase() + "." + hbaseMapping.getTable();
// 拼接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字段
if (params != null && params.size() == 1 && hbaseMapping.getEtlCondition() == null) { Util.sqlRS(ds, sql, rs -> {
AtomicBoolean stExists = new AtomicBoolean(false); try {
// 验证是否有SYS_TIME字段 ResultSetMetaData rsmd = rs.getMetaData();
Util.sqlRS(ds, sql, rs -> { int cnt = rsmd.getColumnCount();
try { for (int i = 1; i <= cnt; i++) {
ResultSetMetaData rsmd = rs.getMetaData(); String columnName = rsmd.getColumnName(i);
int cnt = rsmd.getColumnCount(); if ("SYS_TIME".equalsIgnoreCase(columnName)) {
for (int i = 1; i <= cnt; i++) { stExists.set(true);
String columnName = rsmd.getColumnName(i); break;
if ("SYS_TIME".equalsIgnoreCase(columnName)) { }
stExists.set(true); }
break; } catch (Exception e) {
} // ignore
} }
} catch (Exception e) { return null;
// ignore });
} if (stExists.get()) {
return null; sql += " WHERE SYS_TIME >= '" + params.get(0) + "' ";
}); }
if (stExists.get()) { } else if (hbaseMapping.getEtlCondition() != null && params != null) {
sql += " WHERE SYS_TIME >= '" + params.get(0) + "' "; String etlCondition = hbaseMapping.getEtlCondition();
} int size = params.size();
} else if (hbaseMapping.getEtlCondition() != null && params != null) { for (int i = 0; i < size; i++) {
String etlCondition = hbaseMapping.getEtlCondition(); etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
int size = params.size(); }
for (int i = 0; i < size; i++) {
etlCondition = etlCondition.replace("{" + i + "}", params.get(i)); sql += " " + etlCondition;
} }
sql += " " + etlCondition; // 获取总数
} String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
long cnt = (Long) Util.sqlRS(ds, countSql, rs -> {
// 获取总数 Long count = null;
String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; try {
long cnt = (Long) Util.sqlRS(ds, countSql, rs -> { if (rs.next()) {
Long count = null; count = ((Number) rs.getObject(1)).longValue();
try { }
if (rs.next()) { } catch (Exception e) {
count = ((Number) rs.getObject(1)).longValue(); logger.error(e.getMessage(), e);
} }
} catch (Exception e) { return count == null ? 0 : count;
logger.error(e.getMessage(), e); });
}
return count == null ? 0 : count; // 当大于1万条记录时开启多线程
}); if (cnt >= 10000) {
int threadCount = 3;
// 当大于1万条记录时开启多线程 long perThreadCnt = cnt / threadCount;
if (cnt >= 10000) { ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
int threadCount = 3; for (int i = 0; i < threadCount; i++) {
long perThreadCnt = cnt / threadCount; long offset = i * perThreadCnt;
ExecutorService executor = Executors.newFixedThreadPool(threadCount); Long size = null;
List<Future<Boolean>> futures = new ArrayList<>(threadCount); if (i != threadCount - 1) {
for (int i = 0; i < threadCount; i++) { size = perThreadCnt;
long offset = i * perThreadCnt; }
Long size = null; String sqlFinal;
if (i != threadCount - 1) { if (size != null) {
size = perThreadCnt; sqlFinal = sql + " LIMIT " + offset + "," + size;
} } else {
String sqlFinal; sqlFinal = sql + " LIMIT " + offset + "," + cnt;
if (size != null) { }
sqlFinal = sql + " LIMIT " + offset + "," + size; executor.submit(
} else { () -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg));
sqlFinal = sql + " LIMIT " + offset + "," + cnt; }
}
Future<Boolean> future = executor.submit( executor.shutdown();
() -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg)); while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
futures.add(future); // ignore
} }
} else {
for (Future<Boolean> future : futures) { executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg);
future.get(); }
}
logger.info(hbaseMapping.getHbaseTable() + " etl completed in: "
executor.shutdown(); + (System.currentTimeMillis() - start) / 1000 + "s!");
} else {
executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg); etlResult.setResultMessage("导入HBase表 " + hbaseMapping.getHbaseTable() + " 数据：" + successCount.get() + " 条");
} } catch (Exception e) {
logger.error(e.getMessage(), e);
logger.info(hbaseMapping.getHbaseTable() + " etl completed in: " errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage());
+ (System.currentTimeMillis() - start) / 1000 + "s!"); }
etlResult.setResultMessage("导入HBase表 " + hbaseMapping.getHbaseTable() + " 数据：" + successCount.get() + " 条"); if (errMsg.isEmpty()) {
} catch (Exception e) { etlResult.setSucceeded(true);
logger.error(e.getMessage(), e); } else {
errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
} }
return etlResult;
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
* @param ds * @return
* @param sql */
* @param hbaseMapping private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping,
* @param hbaseTemplate HbaseTemplate hbaseTemplate, AtomicLong successCount, List<String> errMsg) {
* @param successCount try {
* @param errMsg Util.sqlRS(ds, sql, rs -> {
* @return int i = 1;
*/
private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping, try {
HbaseTemplate hbaseTemplate, AtomicLong successCount, List<String> errMsg) { boolean complete = false;
try { List<HRow> rows = new ArrayList<>();
Util.sqlRS(ds, sql, rs -> { String[] rowKeyColumns = null;
int i = 1; if (hbaseMapping.getRowKey() != null) {
rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
try { }
boolean complete = false; while (rs.next()) {
List<HRow> rows = new ArrayList<>(); int cc = rs.getMetaData().getColumnCount();
String[] rowKeyColumns = null; int[] jdbcTypes = new int[cc];
if (hbaseMapping.getRowKey() != null) { Class<?>[] classes = new Class[cc];
rowKeyColumns = hbaseMapping.getRowKey().trim().split(","); for (int j = 1; j <= cc; j++) {
} int jdbcType = rs.getMetaData().getColumnType(j);
while (rs.next()) { jdbcTypes[j - 1] = jdbcType;
int cc = rs.getMetaData().getColumnCount(); classes[j - 1] = JdbcTypeUtil.jdbcType2javaType(jdbcType);
int[] jdbcTypes = new int[cc]; }
Class<?>[] classes = new Class[cc]; HRow row = new HRow();
for (int j = 1; j <= cc; j++) {
int jdbcType = rs.getMetaData().getColumnType(j); if (rowKeyColumns != null) {
jdbcTypes[j - 1] = jdbcType; // 取rowKey字段拼接
classes[j - 1] = JdbcTypeUtil.jdbcType2javaType(jdbcType); StringBuilder rowKeyVale = new StringBuilder();
} for (String rowKeyColumnName : rowKeyColumns) {
HRow row = new HRow(); Object obj = rs.getObject(rowKeyColumnName);
if (obj != null) {
if (rowKeyColumns != null) { rowKeyVale.append(obj.toString());
// 取rowKey字段拼接 }
StringBuilder rowKeyVale = new StringBuilder(); rowKeyVale.append("|");
for (String rowKeyColumnName : rowKeyColumns) { }
Object obj = rs.getObject(rowKeyColumnName); int len = rowKeyVale.length();
if (obj != null) { if (len > 0) {
rowKeyVale.append(obj.toString()); rowKeyVale.delete(len - 1, len);
} }
rowKeyVale.append("|"); row.setRowKey(Bytes.toBytes(rowKeyVale.toString()));
} }
int len = rowKeyVale.length();
if (len > 0) { for (int j = 1; j <= cc; j++) {
rowKeyVale.delete(len - 1, len); String columnName = rs.getMetaData().getColumnName(j);
}
row.setRowKey(Bytes.toBytes(rowKeyVale.toString())); Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]);
} if (val == null) {
continue;
for (int j = 1; j <= cc; j++) { }
String columnName = rs.getMetaData().getColumnName(j);
MappingConfig.ColumnItem columnItem = hbaseMapping.getColumnItems().get(columnName);
Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]); // 没有配置映射
if (val == null) { if (columnItem == null) {
continue; String family = hbaseMapping.getFamily();
} String qualifile = columnName;
if (hbaseMapping.isUppercaseQualifier()) {
MappingConfig.ColumnItem columnItem = hbaseMapping.getColumnItems().get(columnName); qualifile = qualifile.toUpperCase();
// 没有配置映射 }
if (columnItem == null) { if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
String family = hbaseMapping.getFamily(); if (hbaseMapping.getRowKey() == null && j == 1) {
String qualifile = columnName; row.setRowKey(Bytes.toBytes(val.toString()));
if (hbaseMapping.isUppercaseQualifier()) { } else {
qualifile = qualifile.toUpperCase(); row.addCell(family, qualifile, Bytes.toBytes(val.toString()));
} }
if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) { } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
if (hbaseMapping.getRowKey() == null && j == 1) { Type type = Type.getType(classes[j - 1]);
row.setRowKey(Bytes.toBytes(val.toString())); if (hbaseMapping.getRowKey() == null && j == 1) {
} else { row.setRowKey(TypeUtil.toBytes(val, type));
row.addCell(family, qualifile, Bytes.toBytes(val.toString())); } else {
} row.addCell(family, qualifile, TypeUtil.toBytes(val, type));
} else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) { }
Type type = Type.getType(classes[j - 1]); } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
if (hbaseMapping.getRowKey() == null && j == 1) { PhType phType = PhType.getType(classes[j - 1]);
row.setRowKey(TypeUtil.toBytes(val, type)); if (hbaseMapping.getRowKey() == null && j == 1) {
} else { row.setRowKey(PhTypeUtil.toBytes(val, phType));
row.addCell(family, qualifile, TypeUtil.toBytes(val, type)); } else {
} row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType));
} else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) { }
PhType phType = PhType.getType(classes[j - 1]); }
if (hbaseMapping.getRowKey() == null && j == 1) { } else {
row.setRowKey(PhTypeUtil.toBytes(val, phType)); // 如果不需要类型转换
} else { if (columnItem.getType() == null || "".equals(columnItem.getType())) {
row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType)); if (val instanceof java.sql.Date) {
} SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd");
} val = dateFmt.format((Date) val);
} else { } else if (val instanceof Timestamp) {
// 如果不需要类型转换 SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
if (columnItem.getType() == null || "".equals(columnItem.getType())) { val = datetimeFmt.format((Date) val);
if (val instanceof java.sql.Date) { }
SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd");
val = dateFmt.format((Date) val); byte[] valBytes = Bytes.toBytes(val.toString());
} else if (val instanceof Timestamp) { if (columnItem.isRowKey()) {
SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); if (columnItem.getRowKeyLen() != null) {
val = datetimeFmt.format((Date) val); valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
} row.setRowKey(valBytes);
} else {
byte[] valBytes = Bytes.toBytes(val.toString()); row.setRowKey(valBytes);
if (columnItem.isRowKey()) { }
if (columnItem.getRowKeyLen() != null) { } else {
valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val)); row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
row.setRowKey(valBytes); }
} else { } else {
row.setRowKey(valBytes); if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
} byte[] valBytes = Bytes.toBytes(val.toString());
} else { if (columnItem.isRowKey()) {
row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); if (columnItem.getRowKeyLen() != null) {
} valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
} else { }
if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) { row.setRowKey(valBytes);
byte[] valBytes = Bytes.toBytes(val.toString()); } else {
if (columnItem.isRowKey()) { row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
if (columnItem.getRowKeyLen() != null) { }
valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val)); } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
} Type type = Type.getType(columnItem.getType());
row.setRowKey(valBytes); if (columnItem.isRowKey()) {
} else { if (columnItem.getRowKeyLen() != null) {
row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes); String v = limitLenNum(columnItem.getRowKeyLen(), val);
} row.setRowKey(Bytes.toBytes(v));
} else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) { } else {
Type type = Type.getType(columnItem.getType()); row.setRowKey(TypeUtil.toBytes(val, type));
if (columnItem.isRowKey()) { }
if (columnItem.getRowKeyLen() != null) { } else {
String v = limitLenNum(columnItem.getRowKeyLen(), val); row.addCell(columnItem.getFamily(),
row.setRowKey(Bytes.toBytes(v)); columnItem.getQualifier(),
} else { TypeUtil.toBytes(val, type));
row.setRowKey(TypeUtil.toBytes(val, type)); }
} } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
} else { PhType phType = PhType.getType(columnItem.getType());
row.addCell(columnItem.getFamily(), if (columnItem.isRowKey()) {
columnItem.getQualifier(), row.setRowKey(PhTypeUtil.toBytes(val, phType));
TypeUtil.toBytes(val, type)); } else {
} row.addCell(columnItem.getFamily(),
} else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) { columnItem.getQualifier(),
PhType phType = PhType.getType(columnItem.getType()); PhTypeUtil.toBytes(val, phType));
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);
if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空"); rows.clear();
complete = true;
rows.add(row); }
complete = false; i++;
if (i % hbaseMapping.getCommitBatch() == 0 && !rows.isEmpty()) { successCount.incrementAndGet();
hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows); if (logger.isDebugEnabled()) {
rows.clear(); logger.debug("successful import count:" + successCount.get());
complete = true; }
} }
i++;
successCount.incrementAndGet(); if (!complete && !rows.isEmpty()) {
if (logger.isDebugEnabled()) { hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
logger.debug("successful import count:" + successCount.get()); }
}
} } catch (Exception e) {
logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e);
if (!complete && !rows.isEmpty()) { errMsg.add(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage());
hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows); // throw new RuntimeException(e);
} }
return i;
} catch (Exception e) { });
logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e); return true;
errMsg.add(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage()); } catch (Exception e) {
// throw new RuntimeException(e); logger.error(e.getMessage(), e);
} return false;
return i; }
}); }
return true;
} catch (Exception e) { private static String limitLenNum(int len, Object val) {
logger.error(e.getMessage(), e); if (val == null) {
return false; return null;
} }
} if (val instanceof Number) {
return String.format("%0" + len + "d", (Number) ((Number) val).longValue());
private static String limitLenNum(int len, Object val) { } else if (val instanceof String) {
if (val == null) { return String.format("%0" + len + "d", Long.parseLong((String) val));
return null; }
} return null;
if (val instanceof Number) { }
return String.format("%0" + len + "d", (Number) ((Number) val).longValue()); }
} else if (val instanceof String) {
return String.format("%0" + len + "d", Long.parseLong((String) val));
}
return null;
}
}

6
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
View File

@@ -20,9 +20,9 @@ import com.alibaba.otter.canal.client.adapter.support.Dml;
*/ */
public class HbaseSyncService { public class HbaseSyncService {
private Logger logger = LoggerFactory.getLogger(this.getClass()); private static Logger logger = LoggerFactory.getLogger(HbaseSyncService.class);
private HbaseTemplate hbaseTemplate; // HBase操作模板 private HbaseTemplate hbaseTemplate; // HBase操作模板
public HbaseSyncService(HbaseTemplate hbaseTemplate){ public HbaseSyncService(HbaseTemplate hbaseTemplate){
this.hbaseTemplate = hbaseTemplate; this.hbaseTemplate = hbaseTemplate;
@@ -140,7 +140,7 @@ public class HbaseSyncService {
Integer.parseInt((String) entry.getValue())); Integer.parseInt((String) entry.getValue()));
bytes = Bytes.toBytes(v); bytes = Bytes.toBytes(v);
} catch (Exception e) { } catch (Exception e) {
// ignore logger.error(e.getMessage(), e);
} }
} }
} }

4
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
View File

@@ -7,12 +7,10 @@ import java.math.RoundingMode;
import java.sql.Timestamp; import java.sql.Timestamp;
import java.util.Date; import java.util.Date;
import com.alibaba.otter.canal.client.adapter.support.Util;
import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Bytes;
import org.joda.time.DateTime;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.google.common.math.LongMath; import com.google.common.math.LongMath;
import org.joda.time.DateTimeZone;
/** /**
* Phoenix类型转换工具类 * Phoenix类型转换工具类

2
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java
View File

@@ -27,7 +27,7 @@ public class EtlLock {
private static final Map<String, InterProcessMutex> DISTRIBUTED_LOCK = new ConcurrentHashMap<>(); private static final Map<String, InterProcessMutex> DISTRIBUTED_LOCK = new ConcurrentHashMap<>();
private static Mode mode = Mode.LOCAL; private Mode mode = Mode.LOCAL;
@Resource @Resource
private CuratorClient curatorClient; private CuratorClient curatorClient;

8
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java
View File

@@ -34,7 +34,7 @@ public class SyncSwitch {
private static final Map<String, BooleanMutex> DISTRIBUTED_LOCK = new ConcurrentHashMap<>(); private static final Map<String, BooleanMutex> DISTRIBUTED_LOCK = new ConcurrentHashMap<>();
private static Mode mode = Mode.LOCAL; private Mode mode = Mode.LOCAL;
@Resource @Resource
private AdapterCanalConfig adapterCanalConfig; private AdapterCanalConfig adapterCanalConfig;
@@ -165,20 +165,20 @@ public class SyncSwitch {
} }
} }
public Boolean status(String destination) { public boolean status(String destination) {
if (mode == Mode.LOCAL) { if (mode == Mode.LOCAL) {
BooleanMutex mutex = LOCAL_LOCK.get(destination); BooleanMutex mutex = LOCAL_LOCK.get(destination);
if (mutex != null) { if (mutex != null) {
return mutex.state(); return mutex.state();
} else { } else {
return null; return false;
} }
} else { } else {
BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination);
if (mutex != null) { if (mutex != null) {
return mutex.state(); return mutex.state();
} else { } else {
return null; return false;
} }
} }
} }

4
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
View File

@@ -3,7 +3,6 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@@ -17,6 +16,7 @@ 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.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.MessageUtil; import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
@@ -43,7 +43,7 @@ public abstract class AbstractCanalAdapterWorker {
public AbstractCanalAdapterWorker(List<List<OuterAdapter>> canalOuterAdapters){ public AbstractCanalAdapterWorker(List<List<OuterAdapter>> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.groupInnerExecutorService = Util.newFixedThreadPool(canalOuterAdapters.size(), 5000L);
syncSwitch = (SyncSwitch) SpringContext.getBean(SyncSwitch.class); syncSwitch = (SyncSwitch) SpringContext.getBean(SyncSwitch.class);
} }

12
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java
View File

@@ -2,13 +2,13 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.List; import java.util.List;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.OuterAdapter; 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.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
/** /**
@@ -44,12 +44,12 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
protected void process() { protected void process() {
while (!running) { while (!running) {
try { try {
Thread.sleep(1000); Thread.sleep(500);
} catch (InterruptedException e) { } catch (InterruptedException e) {
// ignore // ignore
} }
} }
ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); ExecutorService workerExecutor = Util.newSingleThreadExecutor(5000L);
int retry = canalClientConfig.getRetries() == null int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒
@@ -63,8 +63,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector.subscribe(); connector.subscribe();
logger.info("=============> Subscribe topic: {} succeed <=============", this.topic); logger.info("=============> Subscribe topic: {} succeed <=============", this.topic);
while (running) { while (running) {
Boolean status = syncSwitch.status(canalDestination); boolean status = syncSwitch.status(canalDestination);
if (status != null && !status) { if (!status) {
connector.disconnect(); connector.disconnect();
break; break;
} }
@@ -85,6 +85,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
} }
} }
workerExecutor.shutdown();
try { try {
connector.unsubscribe(); connector.unsubscribe();
} catch (WakeupException e) { } catch (WakeupException e) {

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

@@ -5,7 +5,7 @@ import java.net.SocketAddress;
import java.util.*; import java.util.*;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
@@ -176,40 +176,32 @@ public class CanalAdapterLoader {
public void destroy() { public void destroy() {
if (!canalWorkers.isEmpty()) { if (!canalWorkers.isEmpty()) {
ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size());
List<Future<Boolean>> futures = new ArrayList<>();
for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) { for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) {
futures.add(stopExecutorService.submit(() -> { stopExecutorService.execute(canalAdapterWorker::stop);
canalAdapterWorker.stop();
return true;
}));
} }
futures.forEach(future -> { stopExecutorService.shutdown();
try { try {
future.get(); while (!stopExecutorService.awaitTermination(1, TimeUnit.SECONDS)) {
} catch (Exception e) {
// ignore // ignore
} }
}); } catch (InterruptedException e) {
stopExecutorService.shutdown(); // ignore
}
} }
if (!canalMQWorker.isEmpty()) { if (!canalMQWorker.isEmpty()) {
ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size()); ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size());
List<Future<Boolean>> futures = new ArrayList<>();
for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) { for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) {
futures.add(stopMQWorkerService.submit(() -> { stopMQWorkerService.execute(canalAdapterMQWorker::stop);
canalAdapterMQWorker.stop();
return true;
}));
} }
futures.forEach(future -> { stopMQWorkerService.shutdown();
try { try {
future.get(); while (!stopMQWorkerService.awaitTermination(1, TimeUnit.SECONDS)) {
} catch (Exception e) {
// ignore // ignore
} }
}); } catch (InterruptedException e) {
stopMQWorkerService.shutdown(); // ignore
}
} }
logger.info("All canal adapters destroyed"); logger.info("All canal adapters destroyed");
} }

13
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
View File

@@ -2,12 +2,12 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.List; import java.util.List;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.OuterAdapter; 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.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
/** /**
@@ -49,8 +49,9 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
} }
} }
ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); ExecutorService workerExecutor = Util.newSingleThreadExecutor(5000L);
int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒
while (running) { while (running) {
@@ -62,8 +63,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
connector.subscribe(); connector.subscribe();
logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
while (running) { while (running) {
Boolean status = syncSwitch.status(canalDestination); boolean status = syncSwitch.status(canalDestination);
if (status != null && !status) { if (!status) {
connector.disconnect(); connector.disconnect();
break; break;
} }
@@ -84,6 +85,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
} }
} }
workerExecutor.shutdown();
try { try {
connector.unsubscribe(); connector.unsubscribe();
} catch (WakeupException e) { } catch (WakeupException e) {

2
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
View File

@@ -140,7 +140,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
} }
} }
} catch (Exception e) { } catch (Throwable e) {
logger.error("process error!", e); logger.error("process error!", e);
} finally { } finally {
connector.disconnect(); connector.disconnect();

6
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationConfigMonitor.java
View File

@@ -1,7 +1,9 @@
package com.alibaba.otter.canal.adapter.launcher.monitor; package com.alibaba.otter.canal.adapter.launcher.monitor;
import java.io.File; import java.io.File;
import java.io.FileReader; import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map; import java.util.Map;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
@@ -68,7 +70,7 @@ public class ApplicationConfigMonitor {
super.onFileChange(file); super.onFileChange(file);
try { try {
// 检查yml格式 // 检查yml格式
new Yaml().loadAs(new FileReader(file), Map.class); new Yaml().loadAs(new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8), Map.class);
canalAdapterService.destroy(); canalAdapterService.destroy();

11
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java
View File

@@ -1,6 +1,8 @@
package com.alibaba.otter.canal.adapter.launcher.monitor.remote; package com.alibaba.otter.canal.adapter.launcher.monitor.remote;
import java.io.FileWriter; import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.sql.Connection; import java.sql.Connection;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.SQLException; import java.sql.SQLException;
@@ -35,7 +37,7 @@ public class DbRemoteConfigLoader implements RemoteConfigLoader {
private DruidDataSource dataSource; private DruidDataSource dataSource;
private static volatile long currentConfigTimestamp = 0; private volatile long currentConfigTimestamp = 0;
private Map<String, ConfigItem> remoteAdapterConfigs = new MapMaker().makeMap(); private Map<String, ConfigItem> remoteAdapterConfigs = new MapMaker().makeMap();
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2,
@@ -115,7 +117,10 @@ public class DbRemoteConfigLoader implements RemoteConfigLoader {
* @param content 文件内容 * @param content 文件内容
*/ */
private void overrideLocalCanalConfig(String content) { private void overrideLocalCanalConfig(String content) {
try (FileWriter writer = new FileWriter(CommonUtils.getConfPath() + "application.yml")) {
try (OutputStreamWriter writer = new OutputStreamWriter(
new FileOutputStream(CommonUtils.getConfPath() + "application.yml"),
StandardCharsets.UTF_8)) {
writer.write(content); writer.write(content);
writer.flush(); writer.flush();
} catch (Exception e) { } catch (Exception e) {

13
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java
View File

@@ -1,11 +1,14 @@
package com.alibaba.otter.canal.adapter.launcher.monitor.remote; package com.alibaba.otter.canal.adapter.launcher.monitor.remote;
import com.alibaba.otter.canal.common.utils.CommonUtils; import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.File; import com.alibaba.otter.canal.common.utils.CommonUtils;
import java.io.FileWriter;
/** /**
* 远程配置监听器实现 * 远程配置监听器实现
@@ -35,7 +38,9 @@ public class RemoteAdapterMonitorImpl implements RemoteAdapterMonitor {
} }
} }
String name = configItem.getName(); String name = configItem.getName();
try (FileWriter writer = new FileWriter(confPath + category + "/" + configItem.getName())) { try (OutputStreamWriter writer = new OutputStreamWriter(
new FileOutputStream(confPath + category + "/" + configItem.getName()),
StandardCharsets.UTF_8)) {
writer.write(configItem.getContent()); writer.write(configItem.getContent());
writer.flush(); writer.flush();
logger.info("## Loaded remote adapter config: {}/{}", category, name); logger.info("## Loaded remote adapter config: {}/{}", category, name);

34
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java
View File

@@ -69,16 +69,16 @@ public class CommonRest {
} }
try { try {
Boolean oriSwitchStatus; boolean oriSwitchStatus;
if (destination != null) { if (destination != null) {
oriSwitchStatus = syncSwitch.status(destination); oriSwitchStatus = syncSwitch.status(destination);
if (oriSwitchStatus != null && oriSwitchStatus) { if (oriSwitchStatus) {
syncSwitch.off(destination); syncSwitch.off(destination);
} }
} else { } else {
// task可能为destination，直接锁task // task可能为destination，直接锁task
oriSwitchStatus = syncSwitch.status(task); oriSwitchStatus = syncSwitch.status(task);
if (oriSwitchStatus != null && oriSwitchStatus) { if (oriSwitchStatus) {
syncSwitch.off(task); syncSwitch.off(task);
} }
} }
@@ -89,9 +89,9 @@ public class CommonRest {
} }
return adapter.etl(task, paramArray); return adapter.etl(task, paramArray);
} finally { } finally {
if (destination != null && oriSwitchStatus != null && oriSwitchStatus) { if (destination != null && oriSwitchStatus) {
syncSwitch.on(destination); syncSwitch.on(destination);
} else if (destination == null && oriSwitchStatus != null && oriSwitchStatus) { } else if (destination == null && oriSwitchStatus) {
syncSwitch.on(task); syncSwitch.on(task);
} }
} }
@@ -102,7 +102,7 @@ public class CommonRest {
/** /**
* ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST
* *
* @param type 类型 hbase, es * @param type 类型 hbase, es
* @param task 任务名对应配置文件名 mytest_person2.yml * @param task 任务名对应配置文件名 mytest_person2.yml
* @param params etl where条件参数, 为空全部导入 * @param params etl where条件参数, 为空全部导入
@@ -129,7 +129,7 @@ public class CommonRest {
/** /**
* 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml * 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
* *
* @param type 类型 hbase, es * @param type 类型 hbase, es
* @param task 任务名对应配置文件名 mytest_person2.yml * @param task 任务名对应配置文件名 mytest_person2.yml
* @return * @return
@@ -148,11 +148,11 @@ public class CommonRest {
Set<String> destinations = adapterCanalConfig.DESTINATIONS; Set<String> destinations = adapterCanalConfig.DESTINATIONS;
for (String destination : destinations) { for (String destination : destinations) {
Map<String, String> resMap = new LinkedHashMap<>(); Map<String, String> resMap = new LinkedHashMap<>();
Boolean status = syncSwitch.status(destination); boolean status = syncSwitch.status(destination);
String resStatus = "none"; String resStatus;
if (status != null && status) { if (status) {
resStatus = "on"; resStatus = "on";
} else if (status != null && !status) { } else {
resStatus = "off"; resStatus = "off";
} }
resMap.put("destination", destination); resMap.put("destination", destination);
@@ -164,7 +164,7 @@ public class CommonRest {
/** /**
* 实例同步开关 curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT * 实例同步开关 curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT
* *
* @param destination 实例名称 * @param destination 实例名称
* @param status 开关状态: off on * @param status 开关状态: off on
* @return * @return
@@ -189,17 +189,17 @@ public class CommonRest {
/** /**
* 获取实例开关状态 curl http://127.0.0.1:8081/syncSwitch/example * 获取实例开关状态 curl http://127.0.0.1:8081/syncSwitch/example
* *
* @param destination 实例名称 * @param destination 实例名称
* @return * @return
*/ */
@GetMapping("/syncSwitch/{destination}") @GetMapping("/syncSwitch/{destination}")
public Map<String, String> etl(@PathVariable String destination) { public Map<String, String> etl(@PathVariable String destination) {
Boolean status = syncSwitch.status(destination); boolean status = syncSwitch.status(destination);
String resStatus = "none"; String resStatus;
if (status != null && status) { if (status) {
resStatus = "on"; resStatus = "on";
} else if (status != null && !status) { } else {
resStatus = "off"; resStatus = "off";
} }
Map<String, String> res = new LinkedHashMap<>(); Map<String, String> res = new LinkedHashMap<>();

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

@@ -146,7 +146,7 @@ public class RdbAdapter implements OuterAdapter {
skipDupException); skipDupException);
rdbConfigMonitor = new RdbConfigMonitor(); rdbConfigMonitor = new RdbConfigMonitor();
rdbConfigMonitor.init(configuration.getKey(), this); rdbConfigMonitor.init(configuration.getKey(), this, envProperties);
} }
/** /**

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

@@ -1,10 +1,10 @@
package com.alibaba.otter.canal.client.adapter.rdb.config; package com.alibaba.otter.canal.client.adapter.rdb.config;
import org.apache.commons.lang.StringUtils;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import org.apache.commons.lang.StringUtils;
/** /**
* RDB表映射配置 * RDB表映射配置
* *
@@ -21,7 +21,7 @@ public class MappingConfig {
private String outerAdapterKey; // 对应适配器的key private String outerAdapterKey; // 对应适配器的key
private Boolean concurrent; // 是否并行同步 private boolean concurrent = false; // 是否并行同步
private DbMapping dbMapping; // db映射配置 private DbMapping dbMapping; // db映射配置
@@ -49,11 +49,11 @@ public class MappingConfig {
this.outerAdapterKey = outerAdapterKey; this.outerAdapterKey = outerAdapterKey;
} }
public Boolean getConcurrent() { public boolean getConcurrent() {
return concurrent == null ? false : concurrent; return concurrent;
} }
public void setConcurrent(Boolean concurrent) { public void setConcurrent(boolean concurrent) {
this.concurrent = concurrent; this.concurrent = concurrent;
} }
@@ -87,11 +87,11 @@ public class MappingConfig {
public static class DbMapping { public static class DbMapping {
private Boolean mirrorDb = false; // 是否镜像库 private boolean mirrorDb = false; // 是否镜像库
private String database; // 数据库名或schema名 private String database; // 数据库名或schema名
private String table; // 表名 private String table; // 表名
private Map<String, String> targetPk = new LinkedHashMap<>(); // 目标表主键字段 private Map<String, String> targetPk = new LinkedHashMap<>(); // 目标表主键字段
private Boolean mapAll = false; // 映射所有字段 private boolean mapAll = false; // 映射所有字段
private String targetDb; // 目标库名 private String targetDb; // 目标库名
private String targetTable; // 目标表名 private String targetTable; // 目标表名
private Map<String, String> targetColumns; // 目标表字段映射 private Map<String, String> targetColumns; // 目标表字段映射
@@ -103,11 +103,11 @@ public class MappingConfig {
private Map<String, String> allMapColumns; private Map<String, String> allMapColumns;
public Boolean getMirrorDb() { public boolean getMirrorDb() {
return mirrorDb == null ? false : mirrorDb; return mirrorDb;
} }
public void setMirrorDb(Boolean mirrorDb) { public void setMirrorDb(boolean mirrorDb) {
this.mirrorDb = mirrorDb; this.mirrorDb = mirrorDb;
} }

20
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
View File

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.monitor;
import java.io.File; import java.io.File;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.Properties;
import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
@@ -11,8 +12,8 @@ import org.apache.commons.io.monitor.FileAlterationObserver;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter; import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig;
@@ -29,11 +30,14 @@ public class RdbConfigMonitor {
private RdbAdapter rdbAdapter; private RdbAdapter rdbAdapter;
private Properties envProperties;
private FileAlterationMonitor fileMonitor; private FileAlterationMonitor fileMonitor;
public void init(String key, RdbAdapter rdbAdapter) { public void init(String key, RdbAdapter rdbAdapter, Properties envProperties) {
this.key = key; this.key = key;
this.rdbAdapter = rdbAdapter; this.rdbAdapter = rdbAdapter;
this.envProperties = envProperties;
File confDir = Util.getConfDirPath(adapterName); File confDir = Util.getConfDirPath(adapterName);
try { try {
FileAlterationObserver observer = new FileAlterationObserver(confDir, FileAlterationObserver observer = new FileAlterationObserver(confDir,
@@ -64,7 +68,11 @@ public class RdbConfigMonitor {
try { try {
// 加载新增的配置文件 // 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName()); String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate(); config.validate();
if ((key == null && config.getOuterAdapterKey() == null) if ((key == null && config.getOuterAdapterKey() == null)
|| (key != null && key.equals(config.getOuterAdapterKey()))) { || (key != null && key.equals(config.getOuterAdapterKey()))) {
@@ -90,7 +98,11 @@ public class RdbConfigMonitor {
onFileDelete(file); onFileDelete(file);
return; return;
} }
MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class); MappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate(); config.validate();
if ((key == null && config.getOuterAdapterKey() == null) if ((key == null && config.getOuterAdapterKey() == null)
|| (key != null && key.equals(config.getOuterAdapterKey()))) { || (key != null && key.equals(config.getOuterAdapterKey()))) {

26
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
View File

@@ -6,8 +6,7 @@ import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
@@ -78,8 +77,7 @@ public class RdbEtlService {
if (cnt >= 10000) { if (cnt >= 10000) {
int threadCount = 3; int threadCount = 3;
long perThreadCnt = cnt / threadCount; long perThreadCnt = cnt / threadCount;
ExecutorService executor = Executors.newFixedThreadPool(threadCount); ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
List<Future<Boolean>> futures = new ArrayList<>(threadCount);
for (int i = 0; i < threadCount; i++) { for (int i = 0; i < threadCount; i++) {
long offset = i * perThreadCnt; long offset = i * perThreadCnt;
Long size = null; Long size = null;
@@ -92,16 +90,14 @@ public class RdbEtlService {
} else { } else {
sqlFinal = sql + " LIMIT " + offset + "," + cnt; sqlFinal = sql + " LIMIT " + offset + "," + cnt;
} }
Future<Boolean> future = executor executor
.submit(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); .execute(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg));
futures.add(future);
}
for (Future<Boolean> future : futures) {
future.get();
} }
executor.shutdown(); executor.shutdown();
while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
// ignore
}
} else { } else {
executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg);
} }
@@ -202,7 +198,7 @@ public class RdbEtlService {
StringBuilder insertSql = new StringBuilder(); StringBuilder insertSql = new StringBuilder();
insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" ("); insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" (");
columnsMap columnsMap
.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
int len = insertSql.length(); int len = insertSql.length();
insertSql.delete(len - 1, len).append(") VALUES ("); insertSql.delete(len - 1, len).append(") VALUES (");
@@ -213,16 +209,18 @@ public class RdbEtlService {
len = insertSql.length(); len = insertSql.length();
insertSql.delete(len - 1, len).append(")"); insertSql.delete(len - 1, len).append(")");
try (Connection connTarget = targetDS.getConnection(); try (Connection connTarget = targetDS.getConnection();
PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) { PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
connTarget.setAutoCommit(false); connTarget.setAutoCommit(false);
while (rs.next()) { while (rs.next()) {
completed = false;
pstmt.clearParameters(); pstmt.clearParameters();
// 删除数据 // 删除数据
Map<String, Object> values = new LinkedHashMap<>(); Map<String, Object> values = new LinkedHashMap<>();
StringBuilder deleteSql = new StringBuilder( StringBuilder deleteSql = new StringBuilder(
"DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE "); "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE ");
appendCondition(dbMapping, deleteSql, values, rs); appendCondition(dbMapping, deleteSql, values, rs);
try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
int k = 1; int k = 1;

1013
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

File diff suppressed because it is too large Load Diff

7
client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/DBTest.java
View File

@@ -92,17 +92,16 @@ public class DBTest {
private String clob2Str(Clob clob) { private String clob2Str(Clob clob) {
String content = ""; String content = "";
try { try (Reader is = clob.getCharacterStream(); BufferedReader buff = new BufferedReader(is)) {
Reader is = clob.getCharacterStream();
BufferedReader buff = new BufferedReader(is);
String line = buff.readLine(); String line = buff.readLine();
StringBuffer sb = new StringBuffer(); StringBuilder sb = new StringBuilder();
while (line != null) { while (line != null) {
sb.append(line); sb.append(line);
line = buff.readLine(); line = buff.readLine();
} }
content = sb.toString(); content = sb.toString();
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace();
} }
return content; return content;
} }

2
client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/TestConstant.java
View File

@@ -10,7 +10,7 @@ public class TestConstant {
public final static String jdbcUser = "root"; public final static String jdbcUser = "root";
public final static String jdbcPassword = "121212"; public final static String jdbcPassword = "121212";
public static DruidDataSource dataSource; public final static DruidDataSource dataSource;
static { static {
dataSource = new DruidDataSource(); dataSource = new DruidDataSource();

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
View File

@@ -106,8 +106,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
// stage 3 // stage 3
SequenceBarrier dmlParserSequenceBarrier = disruptorMsgBuffer.newBarrier(simpleParserStage.getSequence()); SequenceBarrier dmlParserSequenceBarrier = disruptorMsgBuffer.newBarrier(simpleParserStage.getSequence());
WorkHandler<MessageEvent>[] workHandlers = new DmlParserStage[parserThreadCount]; WorkHandler<MessageEvent>[] workHandlers = new DmlParserStage[tc];
for (int i = 0; i < parserThreadCount; i++) { for (int i = 0; i < tc; i++) {
workHandlers[i] = new DmlParserStage(); workHandlers[i] = new DmlParserStage();
} }
workerPool = new WorkerPool<MessageEvent>(disruptorMsgBuffer, workerPool = new WorkerPool<MessageEvent>(disruptorMsgBuffer,

5
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -49,6 +49,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
properties.put("max.request.size", kafkaProperties.getMaxRequestSize()); properties.put("max.request.size", kafkaProperties.getMaxRequestSize());
properties.put("buffer.memory", kafkaProperties.getBufferMemory()); properties.put("buffer.memory", kafkaProperties.getBufferMemory());
properties.put("key.serializer", StringSerializer.class.getName()); properties.put("key.serializer", StringSerializer.class.getName());
properties.put("max.in.flight.requests.per.connection", 1);
if (!kafkaProperties.getProperties().isEmpty()) { if (!kafkaProperties.getProperties().isEmpty()) {
properties.putAll(kafkaProperties.getProperties()); properties.putAll(kafkaProperties.getProperties());
@@ -160,7 +161,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
if (record != null) { if (record != null) {
if (kafkaProperties.getTransaction()) { if (kafkaProperties.getTransaction()) {
producer.send(record); producer.send(record).get();
} else { } else {
producer.send(record).get(); producer.send(record).get();
} }
@@ -212,7 +213,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
if (kafkaProperties.getTransaction()) { if (kafkaProperties.getTransaction()) {
producer2.send(record); producer2.send(record);
} else { } else {
producer2.send(record).get(); producer2.send(record);
} }
} }