fix #849: hbase数据同步适配

This commit is contained in:
mcy
2018-08-23 12:02:44 +08:00
parent 47348e2e96
commit 20b0b39b14
39 changed files with 3182 additions and 512 deletions

View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.common</artifactId>
<packaging>jar</packaging>
<name>canal client adapter common module for otter ${project.version}</name>
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.protocol</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.12</version>
</dependency>
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.9.4</version>
</dependency>
</dependencies>
</project>

View File

@@ -1,7 +1,8 @@
package com.alibaba.otter.canal.client.adapter; package com.alibaba.otter.canal.client.adapter;
import com.alibaba.otter.canal.client.support.Dml; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.support.SPI; import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.protocol.Message;
/** /**
* 外部适配器接口 * 外部适配器接口
@@ -11,6 +12,7 @@ import com.alibaba.otter.canal.client.support.SPI;
*/ */
@SPI("logger") @SPI("logger")
public interface CanalOuterAdapter { public interface CanalOuterAdapter {
/** /**
* 外部适配器初始化接口 * 外部适配器初始化接口
* *
@@ -21,10 +23,9 @@ public interface CanalOuterAdapter {
/** /**
* 往适配器中写入数据 * 往适配器中写入数据
* *
* @param dml DML操作数据包 * @param message message数据包
* @return 是否成功
*/ */
Boolean writeOut(Dml dml); void writeOut(Message message);
/** /**
* 外部适配器销毁接口 * 外部适配器销毁接口

View File

@@ -1,6 +1,4 @@
package com.alibaba.otter.canal.client.support; package com.alibaba.otter.canal.client.adapter.support;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapterConfiguration;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
@@ -13,15 +11,16 @@ import java.util.Properties;
* @version 1.0.0 * @version 1.0.0
*/ */
public class CanalClientConfig { public class CanalClientConfig {
private String canalServerHost;
private String zookeeperHosts; private String canalServerHost;
private Properties properties; private String zookeeperHosts;
private String bootstrapServers; private Properties properties;
private List<KafkaTopic> kafkaTopics = new ArrayList<>(); private String bootstrapServers;
private List<KafkaTopic> kafkaTopics = new ArrayList<>();
private List<CanalInstance> canalInstances = new ArrayList<>(); private List<CanalInstance> canalInstances = new ArrayList<>();
@@ -74,7 +73,8 @@ public class CanalClientConfig {
} }
public static class CanalInstance { public static class CanalInstance {
private String instance;
private String instance;
private List<AdapterGroup> adapterGroups; private List<AdapterGroup> adapterGroups;
@@ -98,6 +98,7 @@ public class CanalClientConfig {
} }
public static class AdapterGroup { public static class AdapterGroup {
private List<CanalOuterAdapterConfiguration> outAdapters; private List<CanalOuterAdapterConfiguration> outAdapters;
public List<CanalOuterAdapterConfiguration> getOutAdapters() { public List<CanalOuterAdapterConfiguration> getOutAdapters() {
@@ -110,7 +111,8 @@ public class CanalClientConfig {
} }
public static class KafkaTopic { public static class KafkaTopic {
private String topic;
private String topic;
private List<Group> groups = new ArrayList<>(); private List<Group> groups = new ArrayList<>();
@@ -132,9 +134,10 @@ public class CanalClientConfig {
} }
public static class Group { public static class Group {
private String groupId;
// private List<Adaptor> adapters = new ArrayList<>(); private String groupId;
// private List<Adaptor> adapters = new ArrayList<>();
private List<CanalOuterAdapterConfiguration> outAdapters; private List<CanalOuterAdapterConfiguration> outAdapters;
@@ -154,24 +157,25 @@ public class CanalClientConfig {
this.outAdapters = outAdapters; this.outAdapters = outAdapters;
} }
// public List<Adaptor> getAdapters() { // public List<Adaptor> getAdapters() {
// return adapters; // return adapters;
// } // }
// //
// public void setAdapters(List<Adaptor> adapters) { // public void setAdapters(List<Adaptor> adapters) {
// this.adapters = adapters; // this.adapters = adapters;
// } // }
} }
// public static class Adaptor { // public static class Adaptor {
// private List<CanalOuterAdapterConfiguration> outAdapters; // private List<CanalOuterAdapterConfiguration> outAdapters;
// //
// public List<CanalOuterAdapterConfiguration> getOutAdapters() { // public List<CanalOuterAdapterConfiguration> getOutAdapters() {
// return outAdapters; // return outAdapters;
// } // }
// //
// public void setOutAdapters(List<CanalOuterAdapterConfiguration> outAdapters) { // public void setOutAdapters(List<CanalOuterAdapterConfiguration> outAdapters)
// this.outAdapters = outAdapters; // {
// } // this.outAdapters = outAdapters;
// } // }
// }
} }

View File

@@ -1,4 +1,4 @@
package com.alibaba.otter.canal.client.adapter; package com.alibaba.otter.canal.client.adapter.support;
import java.util.Properties; import java.util.Properties;
@@ -9,13 +9,14 @@ import java.util.Properties;
* @version 1.0.0 * @version 1.0.0
*/ */
public class CanalOuterAdapterConfiguration { public class CanalOuterAdapterConfiguration {
private String name; // 适配器名称, 如: logger, hbase, es
private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址 private String name; // 适配器名称, 如: logger, hbase, es
private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址 private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址
private Properties properties; // 其余参数, 可填写适配器中的所需的配置信息 private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址
private Properties properties; // 其余参数, 可填写适配器中的所需的配置信息
public String getName() { public String getName() {
return name; return name;

View File

@@ -1,5 +1,4 @@
package com.alibaba.otter.canal.client.support; package com.alibaba.otter.canal.client.adapter.support;
import java.io.Serializable; import java.io.Serializable;
import java.util.List; import java.util.List;
@@ -12,26 +11,17 @@ import java.util.Map;
* @version 1.0.0 * @version 1.0.0
*/ */
public class Dml implements Serializable { public class Dml implements Serializable {
private static final long serialVersionUID = 2611556444074013268L;
private String canalDestination; private static final long serialVersionUID = 2611556444074013268L;
private String database;
private String table; private String database;
private String type; private String table;
private Long ts; private String type;
private String sql; private Long ts;
private String sql;
private List<Map<String, Object>> data; private List<Map<String, Object>> data;
private List<Map<String, Object>> old; private List<Map<String, Object>> old;
public String getCanalDestination() {
return canalDestination;
}
public void setCanalDestination(String canalDestination) {
this.canalDestination = canalDestination;
}
public String getDatabase() { public String getDatabase() {
return database; return database;
} }
@@ -89,7 +79,6 @@ public class Dml implements Serializable {
} }
public void clear() { public void clear() {
canalDestination = null;
database = null; database = null;
table = null; table = null;
type = null; type = null;
@@ -101,15 +90,7 @@ public class Dml implements Serializable {
@Override @Override
public String toString() { public String toString() {
return "Dml{" + return "Dml{" + "database='" + database + '\'' + ", table='" + table + '\'' + ", type='" + type + '\'' + ", ts="
"canalDestination='" + canalDestination + '\'' + + ts + ", sql='" + sql + '\'' + ", data=" + data + ", old=" + old + '}';
", database='" + database + '\'' +
", table='" + table + '\'' +
", type='" + type + '\'' +
", ts=" + ts +
", sql='" + sql + '\'' +
", data=" + data +
", old=" + old +
'}';
} }
} }

View File

@@ -1,13 +1,6 @@
package com.alibaba.otter.canal.client.support; package com.alibaba.otter.canal.client.adapter.support;
import org.apache.commons.lang.StringUtils; import java.io.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.File;
import java.io.FilenameFilter;
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;
@@ -17,6 +10,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/** /**
* SPI 类加载器 * SPI 类加载器
* *
@@ -24,33 +20,36 @@ import java.util.regex.Pattern;
* @version 1.0.0 * @version 1.0.0
*/ */
public class ExtensionLoader<T> { public class ExtensionLoader<T> {
private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class);
private static final String SERVICES_DIRECTORY = "META-INF/services/"; private static final Logger logger = LoggerFactory
.getLogger(ExtensionLoader.class);
private static final String CANAL_DIRECTORY = "META-INF/canal/"; private static final String SERVICES_DIRECTORY = "META-INF/services/";
private static final String DEFAULT_CLASSLOADER_POLICY = "internal"; private static final String CANAL_DIRECTORY = "META-INF/canal/";
private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*"); private static final String DEFAULT_CLASSLOADER_POLICY = "internal";
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>(); private static final Pattern NAME_SEPARATOR = Pattern
.compile("\\s*[,]+\\s*");
private static final ConcurrentMap<Class<?>, Object> EXTENSION_INSTANCES = new ConcurrentHashMap<>(); private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
private final Class<?> type; private static final ConcurrentMap<Class<?>, Object> EXTENSION_INSTANCES = new ConcurrentHashMap<>();
private final String classLoaderPolicy; private final Class<?> type;
private final ConcurrentMap<Class<?>, String> cachedNames = new ConcurrentHashMap<>(); private final String classLoaderPolicy;
private final Holder<Map<String, Class<?>>> cachedClasses = new Holder<>(); private final ConcurrentMap<Class<?>, String> cachedNames = new ConcurrentHashMap<>();
private final ConcurrentMap<String, Holder<Object>> cachedInstances = new ConcurrentHashMap<>(); private final Holder<Map<String, Class<?>>> cachedClasses = new Holder<>();
private String cachedDefaultName; private final ConcurrentMap<String, Holder<Object>> cachedInstances = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, IllegalStateException> exceptions = new ConcurrentHashMap<>(); private String cachedDefaultName;
private ConcurrentHashMap<String, IllegalStateException> exceptions = new ConcurrentHashMap<>();
private static <T> boolean withExtensionAnnotation(Class<T> type) { private static <T> boolean withExtensionAnnotation(Class<T> type) {
return type.isAnnotationPresent(SPI.class); return type.isAnnotationPresent(SPI.class);
@@ -62,14 +61,13 @@ public class ExtensionLoader<T> {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type, String classLoaderPolicy) { public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type, String classLoaderPolicy) {
if (type == null) if (type == null) throw new IllegalArgumentException("Extension type == null");
throw new IllegalArgumentException("Extension type == null");
if (!type.isInterface()) { if (!type.isInterface()) {
throw new IllegalArgumentException("Extension type(" + type + ") is not interface!"); throw new IllegalArgumentException("Extension type(" + type + ") is not interface!");
} }
if (!withExtensionAnnotation(type)) { if (!withExtensionAnnotation(type)) {
throw new IllegalArgumentException("Extension type(" + type + throw new IllegalArgumentException("Extension type(" + type + ") is not extension, because WITHOUT @"
") is not extension, because WITHOUT @" + SPI.class.getSimpleName() + " Annotation!"); + SPI.class.getSimpleName() + " Annotation!");
} }
ExtensionLoader<T> loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type); ExtensionLoader<T> loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
@@ -80,7 +78,7 @@ public class ExtensionLoader<T> {
return loader; return loader;
} }
private ExtensionLoader(Class<?> type, String classLoaderPolicy) { private ExtensionLoader(Class<?> type, String classLoaderPolicy){
this.type = type; this.type = type;
this.classLoaderPolicy = classLoaderPolicy; this.classLoaderPolicy = classLoaderPolicy;
} }
@@ -93,7 +91,6 @@ public class ExtensionLoader<T> {
return cachedNames.get(extensionClass); return cachedNames.get(extensionClass);
} }
public ConcurrentHashMap<String, IllegalStateException> getExceptions() { public ConcurrentHashMap<String, IllegalStateException> getExceptions() {
return exceptions; return exceptions;
} }
@@ -107,8 +104,7 @@ public class ExtensionLoader<T> {
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public T getLoadedExtension(String name) { public T getLoadedExtension(String name) {
if (name == null || name.length() == 0) if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null");
throw new IllegalArgumentException("Extension name == null");
Holder<Object> holder = cachedInstances.get(name); Holder<Object> holder = cachedInstances.get(name);
if (holder == null) { if (holder == null) {
cachedInstances.putIfAbsent(name, new Holder<>()); cachedInstances.putIfAbsent(name, new Holder<>());
@@ -136,8 +132,7 @@ public class ExtensionLoader<T> {
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public T getExtension(String name) { public T getExtension(String name) {
if (name == null || name.length() == 0) if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null");
throw new IllegalArgumentException("Extension name == null");
if ("true".equals(name)) { if ("true".equals(name)) {
return getDefaultExtension(); return getDefaultExtension();
} }
@@ -164,16 +159,14 @@ public class ExtensionLoader<T> {
*/ */
public T getDefaultExtension() { public T getDefaultExtension() {
getExtensionClasses(); getExtensionClasses();
if (null == cachedDefaultName || cachedDefaultName.length() == 0 if (null == cachedDefaultName || cachedDefaultName.length() == 0 || "true".equals(cachedDefaultName)) {
|| "true".equals(cachedDefaultName)) {
return null; return null;
} }
return getExtension(cachedDefaultName); return getExtension(cachedDefaultName);
} }
public boolean hasExtension(String name) { public boolean hasExtension(String name) {
if (name == null || name.length() == 0) if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null");
throw new IllegalArgumentException("Extension name == null");
try { try {
return getExtensionClass(name) != null; return getExtensionClass(name) != null;
} catch (Throwable t) { } catch (Throwable t) {
@@ -197,7 +190,7 @@ public class ExtensionLoader<T> {
/** /**
* 编程方式添加新扩展点 * 编程方式添加新扩展点
* *
* @param name 扩展点名 * @param name 扩展点名
* @param clazz 扩展点类 * @param clazz 扩展点类
* @throws IllegalStateException 要添加扩展点名已经存在 * @throws IllegalStateException 要添加扩展点名已经存在
*/ */
@@ -205,20 +198,17 @@ public class ExtensionLoader<T> {
getExtensionClasses(); // load classes getExtensionClasses(); // load classes
if (!type.isAssignableFrom(clazz)) { if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException("Input type " + throw new IllegalStateException("Input type " + clazz + "not implement Extension " + type);
clazz + "not implement Extension " + type);
} }
if (clazz.isInterface()) { if (clazz.isInterface()) {
throw new IllegalStateException("Input type " + throw new IllegalStateException("Input type " + clazz + "can not be interface!");
clazz + "can not be interface!");
} }
if (StringUtils.isBlank(name)) { if (name == null || "".equals(name)) {
throw new IllegalStateException("Extension name is blank (Extension " + type + ")!"); throw new IllegalStateException("Extension name is blank (Extension " + type + ")!");
} }
if (cachedClasses.get().containsKey(name)) { if (cachedClasses.get().containsKey(name)) {
throw new IllegalStateException("Extension name " + throw new IllegalStateException("Extension name " + name + " already existed(Extension " + type + ")!");
name + " already existed(Extension " + type + ")!");
} }
cachedNames.put(clazz, name); cachedNames.put(clazz, name);
@@ -228,7 +218,7 @@ public class ExtensionLoader<T> {
/** /**
* 编程方式添加替换已有扩展点 * 编程方式添加替换已有扩展点
* *
* @param name 扩展点名 * @param name 扩展点名
* @param clazz 扩展点类 * @param clazz 扩展点类
* @throws IllegalStateException 要添加扩展点名已经存在 * @throws IllegalStateException 要添加扩展点名已经存在
* @deprecated 不推荐应用使用,一般只在测试时可以使用 * @deprecated 不推荐应用使用,一般只在测试时可以使用
@@ -238,20 +228,17 @@ public class ExtensionLoader<T> {
getExtensionClasses(); // load classes getExtensionClasses(); // load classes
if (!type.isAssignableFrom(clazz)) { if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException("Input type " + throw new IllegalStateException("Input type " + clazz + "not implement Extension " + type);
clazz + "not implement Extension " + type);
} }
if (clazz.isInterface()) { if (clazz.isInterface()) {
throw new IllegalStateException("Input type " + throw new IllegalStateException("Input type " + clazz + "can not be interface!");
clazz + "can not be interface!");
} }
if (StringUtils.isBlank(name)) { if (name == null || "".equals(name)) {
throw new IllegalStateException("Extension name is blank (Extension " + type + ")!"); throw new IllegalStateException("Extension name is blank (Extension " + type + ")!");
} }
if (!cachedClasses.get().containsKey(name)) { if (!cachedClasses.get().containsKey(name)) {
throw new IllegalStateException("Extension name " + throw new IllegalStateException("Extension name " + name + " not existed(Extension " + type + ")!");
name + " not existed(Extension " + type + ")!");
} }
cachedNames.put(clazz, name); cachedNames.put(clazz, name);
@@ -259,13 +246,12 @@ public class ExtensionLoader<T> {
cachedInstances.remove(name); cachedInstances.remove(name);
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private T createExtension(String name) { private T createExtension(String name) {
Class<?> clazz = getExtensionClasses().get(name); Class<?> clazz = getExtensionClasses().get(name);
if (clazz == null) { if (clazz == null) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
type + ") could not be instantiated: class could not be found"); + ") could not be instantiated: class could not be found");
} }
try { try {
T instance = (T) EXTENSION_INSTANCES.get(clazz); T instance = (T) EXTENSION_INSTANCES.get(clazz);
@@ -275,17 +261,15 @@ public class ExtensionLoader<T> {
} }
return instance; return instance;
} catch (Throwable t) { } catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
type + ") could not be instantiated: " + t.getMessage(), t); + ") could not be instantiated: " + t.getMessage(),
t);
} }
} }
private Class<?> getExtensionClass(String name) { private Class<?> getExtensionClass(String name) {
if (type == null) if (type == null) throw new IllegalArgumentException("Extension type == null");
throw new IllegalArgumentException("Extension type == null"); if (name == null) throw new IllegalArgumentException("Extension name == null");
if (name == null)
throw new IllegalArgumentException("Extension name == null");
Class<?> clazz = getExtensionClasses().get(name); Class<?> clazz = getExtensionClasses().get(name);
if (clazz == null) if (clazz == null)
throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName() + "!"); throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName() + "!");
@@ -312,13 +296,14 @@ public class ExtensionLoader<T> {
throw new IllegalStateException("failed to get class loader resource"); throw new IllegalStateException("failed to get class loader resource");
} }
String dirtyPath = url.toString(); String dirtyPath = url.toString();
String jarPath = dirtyPath.replaceAll("^.*file:/", ""); //removes file:/ and everything before it String jarPath = dirtyPath.replaceAll("^.*file:/", ""); // removes file:/ and everything before it
jarPath = jarPath.replaceAll("jar!.*", "jar"); //removes everything after .jar, if .jar exists in dirtyPath jarPath = jarPath.replaceAll("jar!.*", "jar"); // removes everything after .jar, if .jar exists in dirtyPath
jarPath = jarPath.replaceAll("%20", " "); //necessary if path has spaces within jarPath = jarPath.replaceAll("%20", " "); // necessary if path has spaces within
if (!jarPath.endsWith(".jar")) { // this is needed if you plan to run the app using Spring Tools Suit play button. if (!jarPath.endsWith(".jar")) { // this is needed if you plan to run the app using Spring Tools Suit play
// button.
jarPath = jarPath.replaceAll("/classes/.*", "/classes/"); jarPath = jarPath.replaceAll("/classes/.*", "/classes/");
} }
return Paths.get(jarPath).getParent().toString(); //Paths - from java 8 return Paths.get(jarPath).getParent().toString(); // Paths - from java 8
} }
private Map<String, Class<?>> loadExtensionClasses() { private Map<String, Class<?>> loadExtensionClasses() {
@@ -329,7 +314,7 @@ public class ExtensionLoader<T> {
String[] names = NAME_SEPARATOR.split(value); String[] names = NAME_SEPARATOR.split(value);
if (names.length > 1) { if (names.length > 1) {
throw new IllegalStateException("more than 1 default extension name on extension " + type.getName() throw new IllegalStateException("more than 1 default extension name on extension " + type.getName()
+ ": " + Arrays.toString(names)); + ": " + Arrays.toString(names));
} }
if (names.length == 1) cachedDefaultName = names[0]; if (names.length == 1) cachedDefaultName = names[0];
} }
@@ -337,12 +322,17 @@ public class ExtensionLoader<T> {
Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>(); Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>();
//1. lib folder,customized extension classLoader (jar_dir/lib) // 1. lib folder,customized extension classLoader (jar_dir/lib)
String dir = File.separator + this.getJarDirectoryPath() + File.separator + "lib"; String dir = File.separator + this.getJarDirectoryPath() + File.separator + "lib";
logger.info("extension classpath dir: " + dir); logger.info("extension classpath dir: " + dir);
File externalLibDir = new File(dir); File externalLibDir = new File(dir);
if (!externalLibDir.exists()) {
externalLibDir = new File(
File.separator + this.getJarDirectoryPath() + File.separator + "canal_client" + File.separator + "lib");
}
if (externalLibDir.exists()) { if (externalLibDir.exists()) {
File[] files = externalLibDir.listFiles(new FilenameFilter() { File[] files = externalLibDir.listFiles(new FilenameFilter() {
@Override @Override
public boolean accept(File dir, String name) { public boolean accept(File dir, String name) {
return name.endsWith(".jar"); return name.endsWith(".jar");
@@ -359,8 +349,10 @@ public class ExtensionLoader<T> {
ClassLoader parent = Thread.currentThread().getContextClassLoader(); ClassLoader parent = Thread.currentThread().getContextClassLoader();
URLClassLoader localClassLoader; URLClassLoader localClassLoader;
if (!StringUtils.isEmpty(classLoaderPolicy) && DEFAULT_CLASSLOADER_POLICY.equalsIgnoreCase(classLoaderPolicy)) { if (classLoaderPolicy == null || "".equals(classLoaderPolicy)
localClassLoader = new URLClassLoader(new URL[]{url}, parent) { || DEFAULT_CLASSLOADER_POLICY.equalsIgnoreCase(classLoaderPolicy)) {
localClassLoader = new URLClassLoader(new URL[] { url }, parent) {
@Override @Override
public Class<?> loadClass(String name) throws ClassNotFoundException { public Class<?> loadClass(String name) throws ClassNotFoundException {
Class<?> c = findLoadedClass(name); Class<?> c = findLoadedClass(name);
@@ -368,15 +360,13 @@ public class ExtensionLoader<T> {
return c; return c;
} }
if (name.startsWith("java.") if (name.startsWith("java.") || name.startsWith("org.slf4j.")
|| name.startsWith("org.slf4j.") || name.startsWith("org.apache.logging")
|| name.startsWith("org.apache.logging") || name.startsWith("org.apache.commons.logging.")) {
|| name.startsWith("org.apache.commons.logging.")) { // || name.startsWith("org.apache.hadoop.")) {
//|| name.startsWith("org.apache.hadoop.")) {
c = super.loadClass(name); c = super.loadClass(name);
} }
if (c != null) if (c != null) return c;
return c;
try { try {
// 先加载jar内的class,可避免jar冲突 // 先加载jar内的class,可避免jar冲突
@@ -391,19 +381,19 @@ public class ExtensionLoader<T> {
return super.loadClass(name); return super.loadClass(name);
} }
// @Override @Override
// public Enumeration<URL> getResources(String name) throws IOException { public Enumeration<URL> getResources(String name) throws IOException {
// @SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
// Enumeration<URL>[] tmp = (Enumeration<URL>[]) new Enumeration<?>[2]; Enumeration<URL>[] tmp = (Enumeration<URL>[]) new Enumeration<?>[2];
// tmp[0] = findResources(name); //local class path first tmp[0] = findResources(name); // local class path first
// //tmp[1] = super.getResources(name); // tmp[1] = super.getResources(name);
// return new CompoundEnumeration<>(tmp); return new CompoundEnumeration<>(tmp);
// } }
}; };
} else { } else {
localClassLoader = new URLClassLoader(new URL[]{url}, parent); localClassLoader = new URLClassLoader(new URL[] { url }, parent);
} }
loadFile(extensionClasses, CANAL_DIRECTORY, localClassLoader); loadFile(extensionClasses, CANAL_DIRECTORY, localClassLoader);
@@ -411,7 +401,7 @@ public class ExtensionLoader<T> {
} }
} }
} }
//2. load inner extension class with default classLoader // 2. load inner extension class with default classLoader
ClassLoader classLoader = findClassLoader(); ClassLoader classLoader = findClassLoader();
loadFile(extensionClasses, CANAL_DIRECTORY, classLoader); loadFile(extensionClasses, CANAL_DIRECTORY, classLoader);
loadFile(extensionClasses, SERVICES_DIRECTORY, classLoader); loadFile(extensionClasses, SERVICES_DIRECTORY, classLoader);
@@ -419,6 +409,40 @@ public class ExtensionLoader<T> {
return extensionClasses; return extensionClasses;
} }
public static class CompoundEnumeration<E> implements Enumeration<E> {
private Enumeration<E>[] enums;
private int index = 0;
public CompoundEnumeration(Enumeration<E>[] enums){
this.enums = enums;
}
private boolean next() {
while (this.index < this.enums.length) {
if (this.enums[this.index] != null && this.enums[this.index].hasMoreElements()) {
return true;
}
++this.index;
}
return false;
}
public boolean hasMoreElements() {
return this.next();
}
public E nextElement() {
if (!this.next()) {
throw new NoSuchElementException();
} else {
return this.enums[this.index].nextElement();
}
}
}
private void loadFile(Map<String, Class<?>> extensionClasses, String dir, ClassLoader classLoader) { private void loadFile(Map<String, Class<?>> extensionClasses, String dir, ClassLoader classLoader) {
String fileName = dir + type.getName(); String fileName = dir + type.getName();
try { try {
@@ -450,11 +474,13 @@ public class ExtensionLoader<T> {
} }
if (line.length() > 0) { if (line.length() > 0) {
Class<?> clazz = classLoader.loadClass(line); Class<?> clazz = classLoader.loadClass(line);
//Class<?> clazz = Class.forName(line, true, classLoader); // Class<?> clazz = Class.forName(line, true, classLoader);
if (!type.isAssignableFrom(clazz)) { if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException("Error when load extension class(interface: " + throw new IllegalStateException(
type + ", class line: " + clazz.getName() + "), class " "Error when load extension class(interface: " + type
+ clazz.getName() + "is not subtype of interface."); + ", class line: " + clazz.getName()
+ "), class " + clazz.getName()
+ "is not subtype of interface.");
} else { } else {
try { try {
clazz.getConstructor(type); clazz.getConstructor(type);
@@ -471,7 +497,11 @@ 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 " + type.getName() + " name " + n + " on " + c.getName() + " and " + clazz.getName()); throw new IllegalStateException(
"Duplicate extension " + type.getName() + " name "
+ n + " on "
+ c.getName() + " and "
+ clazz.getName());
} }
} }
} }
@@ -479,7 +509,12 @@ public class ExtensionLoader<T> {
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: " + type + ", class line: " + line + ") in " + url + ", cause: " + t.getMessage(), t); IllegalStateException e = new IllegalStateException(
"Failed to load extension class(interface: " + type + ", class line: "
+ line + ") in " + url
+ ", cause: "
+ t.getMessage(),
t);
exceptions.put(line, e); exceptions.put(line, e);
} }
} }
@@ -490,18 +525,19 @@ public class ExtensionLoader<T> {
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + logger.error("Exception when load extension class(interface: " + type + ", class file: " + url
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: " + logger.error(
type + ", description file: " + fileName + ").", t); "Exception when load extension class(interface: " + type + ", description file: " + fileName + ").",
t);
} }
} }
private static ClassLoader findClassLoader() { private static ClassLoader findClassLoader() {
return ExtensionLoader.class.getClassLoader(); return ExtensionLoader.class.getClassLoader();
} }

View File

@@ -1,8 +1,4 @@
package com.alibaba.otter.canal.client.support; package com.alibaba.otter.canal.client.adapter.support;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.math.BigInteger; import java.math.BigInteger;
@@ -11,49 +7,19 @@ import java.sql.Time;
import java.sql.Timestamp; import java.sql.Timestamp;
import java.sql.Types; import java.sql.Types;
public class JdbcTypeUtil { import org.joda.time.DateTime;
private static Logger logger = LoggerFactory.getLogger(JdbcTypeUtil.class); import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public static Class<?> jdbcType2javaType(int jdbcType) { /**
switch (jdbcType) { * 类型转换工具类
case Types.BIT: *
case Types.BOOLEAN: * @author machengyuan 2018-8-19 下午06:14:23
// return Boolean.class; * @version 1.0.0
case Types.TINYINT: */
return Byte.TYPE; public class JdbcTypeUtil {
case Types.SMALLINT:
return Short.class; private static Logger logger = LoggerFactory.getLogger(JdbcTypeUtil.class);
case Types.INTEGER:
return Integer.class;
case Types.BIGINT:
return Long.class;
case Types.DECIMAL:
case Types.NUMERIC:
return BigDecimal.class;
case Types.REAL:
return Float.class;
case Types.FLOAT:
case Types.DOUBLE:
return Double.class;
case Types.CHAR:
case Types.VARCHAR:
case Types.LONGVARCHAR:
return String.class;
case Types.BINARY:
case Types.VARBINARY:
case Types.LONGVARBINARY:
case Types.BLOB:
return byte[].class;
case Types.DATE:
return Date.class;
case Types.TIME:
return Time.class;
case Types.TIMESTAMP:
return Timestamp.class;
default:
return String.class;
}
}
public static Object typeConvert(String tableName, String columnName, String value, int sqlType, String mysqlType) { public static Object typeConvert(String tableName, String columnName, String value, int sqlType, String mysqlType) {
if (value == null || value.equals("")) { if (value == null || value.equals("")) {
@@ -80,7 +46,7 @@ public class JdbcTypeUtil {
res = Long.parseLong(value); res = Long.parseLong(value);
} }
break; break;
//case Types.BIT: // case Types.BIT:
case Types.BOOLEAN: case Types.BOOLEAN:
res = !"0".equals(value); res = !"0".equals(value);
break; break;
@@ -130,7 +96,7 @@ public class JdbcTypeUtil {
} }
return res; return res;
} catch (Exception e) { } catch (Exception e) {
logger.error("table: {} column: {}, failed convert type {} to {}", tableName, columnName,value, sqlType); logger.error("table: {} column: {}, failed convert type {} to {}", tableName, columnName, value, sqlType);
return value; return value;
} }
} }

View File

@@ -0,0 +1,113 @@
package com.alibaba.otter.canal.client.adapter.support;
import java.util.*;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
/**
* Message对象解析工具类
*
* @author machengyuan 2018-8-19 下午06:14:23
* @version 1.0.0
*/
public class MessageUtil {
public static void parse4Dml(Message message, Consumer<Dml> consumer) {
if (message == null) {
return;
}
List<CanalEntry.Entry> entries = message.getEntries();
for (CanalEntry.Entry entry : entries) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
|| entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
CanalEntry.RowChange rowChange;
try {
rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
CanalEntry.EventType eventType = rowChange.getEventType();
final Dml dml = new Dml();
dml.setDatabase(entry.getHeader().getSchemaName());
dml.setTable(entry.getHeader().getTableName());
dml.setType(eventType.toString());
dml.setTs(System.currentTimeMillis());
dml.setSql(rowChange.getSql());
List<Map<String, Object>> data = new ArrayList<>();
List<Map<String, Object>> old = new ArrayList<>();
if (!rowChange.getIsDdl()) {
Set<String> updateSet = new HashSet<>();
for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE
&& eventType != CanalEntry.EventType.DELETE) {
continue;
}
Map<String, Object> row = new LinkedHashMap<>();
List<CanalEntry.Column> columns;
if (eventType == CanalEntry.EventType.DELETE) {
columns = rowData.getBeforeColumnsList();
} else {
columns = rowData.getAfterColumnsList();
}
for (CanalEntry.Column column : columns) {
row.put(column.getName(),
JdbcTypeUtil.typeConvert(dml.getTable(),
column.getName(),
column.getValue(),
column.getSqlType(),
column.getMysqlType()));
// 获取update为true的字段
if (column.getUpdated()) {
updateSet.add(column.getName());
}
}
if (!row.isEmpty()) {
data.add(row);
}
if (eventType == CanalEntry.EventType.UPDATE) {
Map<String, Object> rowOld = new LinkedHashMap<>();
for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
if (updateSet.contains(column.getName())) {
rowOld.put(column.getName(),
JdbcTypeUtil.typeConvert(dml.getTable(),
column.getName(),
column.getValue(),
column.getSqlType(),
column.getMysqlType()));
}
}
// update操作将记录修改前的值
if (!rowOld.isEmpty()) {
old.add(rowOld);
}
}
}
if (!data.isEmpty()) {
dml.setData(data);
}
if (!old.isEmpty()) {
dml.setOld(old);
}
consumer.accept(dml);
}
}
}
public interface Consumer<T> {
void accept(T t);
}
}

View File

@@ -1,11 +1,12 @@
package com.alibaba.otter.canal.client.support; package com.alibaba.otter.canal.client.adapter.support;
import java.lang.annotation.*; import java.lang.annotation.*;
@Documented @Documented
@Retention(RetentionPolicy.RUNTIME) @Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE}) @Target({ ElementType.TYPE })
public @interface SPI { public @interface SPI {
//Default SPI name
// Default SPI name
String value() default ""; String value() default "";
} }

View File

@@ -0,0 +1,101 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.hbase</artifactId>
<packaging>jar</packaging>
<name>canal client adapter hbase module for otter ${project.version}</name>
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
<version>1.17</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>1.1.2</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>12.0.1</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<archive>
<addMavenDescriptor>true</addMavenDescriptor>
</archive>
<excludes>
<exclude>**/hbase-mapping/**</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<copy todir="${project.basedir}/../../client-launcher/target/canal_client/conf/hbase-mapping" overwrite="true" >
<fileset dir="${project.basedir}/src/main/resources/hbase-mapping" erroronmissingdir="true">
<include name="*.conf"/>
<include name="*.yml"/>
</fileset>
</copy>
</tasks>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

View File

@@ -0,0 +1,115 @@
package com.alibaba.otter.canal.client.adapter.hbase;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader;
import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService;
import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.protocol.Message;
/**
* HBase外部适配器
*
* @author machengyuan 2018-8-21 下午8:45:38
* @version 1.0.0
*/
@SPI("hbase")
public class HbaseAdapter implements CanalOuterAdapter {
private static volatile Map<String, MappingConfig> mappingConfigCache = null;
private Connection conn;
private HbaseSyncService hbaseSyncService;
@Override
public void init(CanalOuterAdapterConfiguration configuration) {
try {
if (mappingConfigCache == null) {
synchronized (MappingConfig.class) {
if (mappingConfigCache == null) {
Map<String, MappingConfig> hbaseMapping = MappingConfigLoader.load();
mappingConfigCache = new HashMap<>();
for (MappingConfig mappingConfig : hbaseMapping.values()) {
mappingConfigCache.put(mappingConfig.getHbaseOrm().getDatabase() + "-"
+ mappingConfig.getHbaseOrm().getTable(),
mappingConfig);
}
}
}
}
String hosts = configuration.getZkHosts();
if (StringUtils.isEmpty(hosts)) {
hosts = configuration.getHosts();
}
if (StringUtils.isEmpty(hosts)) {
throw new RuntimeException("Empty zookeeper hosts");
}
String[] zkHosts = StringUtils.split(hosts, ",");
int zkPort = 0;
StringBuilder hostsWithoutPort = new StringBuilder();
for (String host : zkHosts) {
int i = host.indexOf(":");
hostsWithoutPort.append(host, 0, i);
hostsWithoutPort.append(",");
if (zkPort == 0) zkPort = Integer.parseInt(host.substring(i + 1));
}
hostsWithoutPort.deleteCharAt(hostsWithoutPort.length() - 1);
String znode = configuration.getProperties().getProperty("znodeParent");
if (StringUtils.isEmpty(znode)) {
znode = "/hbase";
}
Configuration hbaseConfig = HBaseConfiguration.create();
hbaseConfig.set("hbase.zookeeper.quorum", hostsWithoutPort.toString());
hbaseConfig.set("hbase.zookeeper.property.clientPort", Integer.toString(zkPort));
hbaseConfig.set("zookeeper.znode.parent", znode);
conn = ConnectionFactory.createConnection(hbaseConfig);
hbaseSyncService = new HbaseSyncService(conn);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@Override
public void writeOut(Message message) {
MessageUtil.parse4Dml(message, new MessageUtil.Consumer<Dml>() {
@Override
public void accept(Dml dml) {
if (dml == null) {
return;
}
String database = dml.getDatabase();
String table = dml.getTable();
MappingConfig config = mappingConfigCache.get(database + "-" + table);
hbaseSyncService.sync(config, dml);
}
});
}
@Override
public void destroy() {
if (conn != null) {
try {
conn.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
}

View File

@@ -0,0 +1,337 @@
package com.alibaba.otter.canal.client.adapter.hbase.config;
import java.util.*;
/**
* HBase表映射配置
*
* @author machengyuan 2018-8-21 下午06:45:49
* @version 1.0.0
*/
public class MappingConfig {
private HbaseOrm hbaseOrm;
public HbaseOrm getHbaseOrm() {
return hbaseOrm;
}
public void setHbaseOrm(HbaseOrm hbaseOrm) {
this.hbaseOrm = hbaseOrm;
}
public void validate() {
if (hbaseOrm.database == null || hbaseOrm.database.isEmpty()) {
throw new NullPointerException("hbaseOrm.database");
}
if (hbaseOrm.table == null || hbaseOrm.table.isEmpty()) {
throw new NullPointerException("hbaseOrm.table");
}
if (hbaseOrm.hbaseTable == null || hbaseOrm.hbaseTable.isEmpty()) {
throw new NullPointerException("hbaseOrm.hbaseTable");
}
if (hbaseOrm.mode == null) {
throw new NullPointerException("hbaseOrm.mode");
}
if (hbaseOrm.rowKey != null && hbaseOrm.rowKeyColumn != null) {
throw new RuntimeException("已配置了复合主键作为RowKey,无需再指定RowKey列");
}
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
MappingConfig config = (MappingConfig) o;
return hbaseOrm != null ? hbaseOrm.equals(config.hbaseOrm) : config.hbaseOrm == null;
}
@Override
public int hashCode() {
return hbaseOrm != null ? hbaseOrm.hashCode() : 0;
}
public static class ColumnItem {
private boolean isRowKey = false;
private String column;
private String family;
private String qualifier;
private String type;
public boolean isRowKey() {
return isRowKey;
}
public void setRowKey(boolean rowKey) {
isRowKey = rowKey;
}
public String getColumn() {
return column;
}
public void setColumn(String column) {
this.column = column;
}
public String getFamily() {
return family;
}
public void setFamily(String family) {
this.family = family;
}
public String getQualifier() {
return qualifier;
}
public void setQualifier(String qualifier) {
this.qualifier = qualifier;
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ColumnItem that = (ColumnItem) o;
return Objects.equals(column, that.column);
}
@Override
public int hashCode() {
return Objects.hash(column);
}
}
public enum Mode {
STRING("STRING"), NATIVE("NATIVE"), PHOENIX("PHOENIX");
private String type;
public String getType() {
return type;
}
Mode(String type){
this.type = type;
}
}
public static class HbaseOrm {
private Mode mode = Mode.STRING;
private String database;
private String table;
private String hbaseTable;
private String family = "CF";
private boolean uppercaseQualifier = true;
private boolean autoCreateTable = false; // 同步时HBase中表不存在的情况下自动建表
private String rowKey; // 指定复合主键为rowKey
private Map<String, String> columns;
private ColumnItem rowKeyColumn;
private String etlCondition;
private Map<String, ColumnItem> columnItems = new LinkedHashMap<>();
private Set<String> families = new LinkedHashSet<>();
private int readBatch = 5000;
private int commitBatch = 5000;
public Mode getMode() {
return mode;
}
public void setMode(Mode mode) {
this.mode = mode;
}
public String getDatabase() {
return database;
}
public void setDatabase(String database) {
this.database = database;
}
public String getTable() {
return table;
}
public void setTable(String table) {
this.table = table;
}
public String getHbaseTable() {
return hbaseTable;
}
public void setHbaseTable(String hbaseTable) {
this.hbaseTable = hbaseTable;
}
public Map<String, String> getColumns() {
return columns;
}
public boolean isAutoCreateTable() {
return autoCreateTable;
}
public void setAutoCreateTable(boolean autoCreateTable) {
this.autoCreateTable = autoCreateTable;
}
public int getReadBatch() {
return readBatch;
}
public void setReadBatch(int readBatch) {
this.readBatch = readBatch;
}
public int getCommitBatch() {
return commitBatch;
}
public void setCommitBatch(int commitBatch) {
this.commitBatch = commitBatch;
}
public String getRowKey() {
return rowKey;
}
public void setRowKey(String rowKey) {
this.rowKey = rowKey;
}
public String getEtlCondition() {
return etlCondition;
}
public void setEtlCondition(String etlCondition) {
this.etlCondition = etlCondition;
}
public void setColumns(Map<String, String> columns) {
this.columns = columns;
if (columns != null) {
for (Map.Entry<String, String> columnField : columns.entrySet()) {
String field = columnField.getValue();
String type = null;
if (field != null) {
// 解析类型
int i = field.indexOf("$");
if (i > -1) {
type = field.substring(i + 1);
field = field.substring(0, i);
}
}
ColumnItem columnItem = new ColumnItem();
columnItem.setColumn(columnField.getKey());
columnItem.setType(type);
if ("rowKey".equalsIgnoreCase(field)) {
columnItem.setRowKey(true);
rowKeyColumn = columnItem;
} else {
if (field == null || field.equals("")) {
columnItem.setFamily(family);
columnItem.setQualifier(columnField.getKey());
} else {
int len = field.indexOf(":");
if (len > -1) {
columnItem.setFamily(field.substring(0, len));
columnItem.setQualifier(field.substring(len + 1));
} else {
columnItem.setFamily(family);
columnItem.setQualifier(field);
}
}
if (uppercaseQualifier) {
columnItem.setQualifier(columnItem.getQualifier().toUpperCase());
}
families.add(columnItem.getFamily());
}
columnItems.put(columnField.getKey(), columnItem);
}
} else {
this.columns = new LinkedHashMap<>();
}
}
public String getFamily() {
return family;
}
public void setFamily(String family) {
this.family = family;
if (family == null) {
this.family = "CF";
}
}
public boolean isUppercaseQualifier() {
return uppercaseQualifier;
}
public void setUppercaseQualifier(boolean uppercaseQualifier) {
this.uppercaseQualifier = uppercaseQualifier;
}
public ColumnItem getRowKeyColumn() {
return rowKeyColumn;
}
public void setRowKeyColumn(ColumnItem rowKeyColumn) {
this.rowKeyColumn = rowKeyColumn;
}
public Map<String, ColumnItem> getColumnItems() {
return columnItems;
}
public void setColumnItems(Map<String, ColumnItem> columnItems) {
this.columnItems = columnItems;
}
public Set<String> getFamilies() {
return families;
}
public void setFamilies(Set<String> families) {
this.families = families;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
HbaseOrm hbaseOrm = (HbaseOrm) o;
if (table != null ? !table.equals(hbaseOrm.table) : hbaseOrm.table != null) return false;
return hbaseTable != null ? hbaseTable.equals(hbaseOrm.hbaseTable) : hbaseOrm.hbaseTable == null;
}
@Override
public int hashCode() {
int result = table != null ? table.hashCode() : 0;
result = 31 * result + (hbaseTable != null ? hbaseTable.hashCode() : 0);
return result;
}
}
}

View File

@@ -0,0 +1,149 @@
package com.alibaba.otter.canal.client.adapter.hbase.config;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
/**
* HBase表映射配置加载器
* <p>
* 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。
* 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式
* </p>
*
* @author machengyuan 2018-8-21 下午06:45:49
* @version 1.0.0
*/
public class MappingConfigLoader {
private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class);
private static final String BASE_PATH = "hbase-mapping/";
/**
* 加载HBase表映射配置
*
* @return 配置名/配置文件名--对象
*/
public static Map<String, MappingConfig> load() {
logger.info("## Start loading mapping config ... ");
String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf");
Map<String, MappingConfig> result = new LinkedHashMap<>();
String[] configLines = mappingConfigContent.split("\n");
for (String c : configLines) {
if (c == null) {
continue;
}
c = c.trim();
if (c.startsWith("#")) {
continue;
}
MappingConfig config;
String configContent = null;
if (c.endsWith(".yml")) {
configContent = readConfigContent(BASE_PATH + "/" + c);
}
// 简单配置database.table@datasourcekey?rowKey=key1,key2
if (StringUtils.isEmpty(configContent)) {
String[] mapping = c.split("\\?");
String params = mapping.length == 2 ? mapping[1] : null;
String rowKey = null;
String srcMeta = mapping[0];
//
if (params != null) {
for (String entry : params.split("&")) {
if ("rowKey".equals(entry.split("=")[0])) {
rowKey = entry.split("=")[1];
}
}
}
String dsKey = srcMeta.split("@").length == 2 ? srcMeta.split("@")[1] : null;
String[] dbTable;
if (dsKey == null) {
dbTable = srcMeta.split("\\.");
} else {
dbTable = srcMeta.split("@")[0].split("\\.");
}
if (dbTable.length == 2) {
config = new MappingConfig();
MappingConfig.HbaseOrm hbaseOrm = new MappingConfig.HbaseOrm();
hbaseOrm.setHbaseTable(dbTable[0].toUpperCase() + "." + dbTable[1].toUpperCase());
hbaseOrm.setAutoCreateTable(true);
hbaseOrm.setDatabase(dbTable[0]);
hbaseOrm.setTable(dbTable[1]);
hbaseOrm.setMode(MappingConfig.Mode.STRING);
hbaseOrm.setRowKey(rowKey);
// 有定义rowKey
if (rowKey != null) {
MappingConfig.ColumnItem columnItem = new MappingConfig.ColumnItem();
columnItem.setRowKey(true);
columnItem.setColumn(rowKey);
hbaseOrm.setRowKeyColumn(columnItem);
}
config.setHbaseOrm(hbaseOrm);
} else {
throw new RuntimeException(String.format("配置项[%s]内容为空, 或格式不符合database.table", c));
}
} else { // 配置文件配置
config = new Yaml().loadAs(configContent, MappingConfig.class);
}
try {
config.validate();
} catch (Exception e) {
throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e);
}
result.put(c, config);
}
logger.info("## Mapping config loaded");
return result;
}
public static String readConfigContent(String config) {
InputStream in = null;
try {
// 先取本地文件,再取类路径
File configFile = new File("config/" + config);
if (configFile.exists()) {
in = new FileInputStream(configFile);
} else {
in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config);
}
if (in == null) {
throw new RuntimeException("Config file not found.");
}
byte[] bytes = new byte[in.available()];
in.read(bytes);
return new String(bytes, "UTF-8");
} catch (IOException e) {
throw new RuntimeException("Read ds-config.yml or hbase-mappings.conf error. ", e);
} finally {
try {
if (in != null) {
in.close();
}
} catch (IOException e) {
// ignore
}
}
}
}

View File

@@ -0,0 +1,418 @@
package com.alibaba.otter.canal.client.adapter.hbase.service;
import java.util.*;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.hbase.support.*;
import com.alibaba.otter.canal.client.adapter.support.Dml;
/**
* HBase同步操作业务
*
* @author machengyuan 2018-8-21 下午06:45:49
* @version 1.0.0
*/
public class HbaseSyncService {
private Logger logger = LoggerFactory.getLogger(this.getClass());
private HbaseTemplate hbaseTemplate; // HBase操作模板
public HbaseSyncService(Connection conn){
hbaseTemplate = new HbaseTemplate(conn);
}
public void sync(MappingConfig config, Dml dml) {
try {
if (config != null) {
String type = dml.getType();
if (type != null && type.equalsIgnoreCase("INSERT")) {
insert(config, dml);
} else if (type != null && type.equalsIgnoreCase("UPDATE")) {
update(config, dml);
} else if (type != null && type.equalsIgnoreCase("DELETE")) {
delete(config, dml);
}
if (logger.isDebugEnabled()) {
String res = dml.toString();
logger.debug(res);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**
* 插入操作
*
* @param config 配置项
* @param dml DML数据
*/
private void insert(MappingConfig config, Dml dml) {
List<Map<String, Object>> data = dml.getData();
if (data == null || data.isEmpty()) {
return;
}
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
// if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable());
// return;
// }
int i = 1;
boolean complete = false;
List<HRow> rows = new ArrayList<>();
for (Map<String, Object> r : data) {
HRow hRow = new HRow();
// 拼接复合rowKey
if (hbaseOrm.getRowKey() != null) {
String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
String rowKeyVale = getRowKeys(rowKeyColumns, r);
// params.put("rowKey", Bytes.toBytes(rowKeyVale));
hRow.setRowKey(Bytes.toBytes(rowKeyVale));
}
convertData2Row(hbaseOrm, hRow, r);
if (hRow.getRowKey() == null) {
throw new RuntimeException("empty rowKey");
}
rows.add(hRow);
complete = false;
if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
rows.clear();
complete = true;
}
i++;
}
if (!complete && !rows.isEmpty()) {
hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
}
}
/**
* 将Map数据转换为HRow行数据
*
* @param hbaseOrm hbase映射配置
* @param hRow 行对象
* @param data Map数据
*/
private static void convertData2Row(MappingConfig.HbaseOrm hbaseOrm, HRow hRow, Map<String, Object> data) {
Map<String, MappingConfig.ColumnItem> columnItems = hbaseOrm.getColumnItems();
int i = 0;
for (Map.Entry<String, Object> entry : data.entrySet()) {
if (entry.getValue() != null) {
MappingConfig.ColumnItem columnItem = columnItems.get(entry.getKey());
byte[] bytes = typeConvert(columnItem, hbaseOrm, entry.getValue());
if (columnItem == null) {
String familyName = hbaseOrm.getFamily();
String qualifier = entry.getKey();
if (hbaseOrm.isUppercaseQualifier()) {
qualifier = qualifier.toUpperCase();
}
if (hbaseOrm.getRowKey() == null && i == 0) {
hRow.setRowKey(bytes);
} else {
hRow.addCell(familyName, qualifier, bytes);
}
} else {
if (columnItem.isRowKey()) {
// row.put("rowKey", bytes);
hRow.setRowKey(bytes);
} else {
hRow.addCell(columnItem.getFamily(), columnItem.getQualifier(), bytes);
}
}
}
i++;
}
}
/**
* 更新操作
*
* @param config
* @param dml
*/
private void update(MappingConfig config, Dml dml) {
List<Map<String, Object>> data = dml.getData();
List<Map<String, Object>> old = dml.getOld();
if (old == null || old.isEmpty() || data == null || data.isEmpty()) {
return;
}
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
// if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable());
// return;
// }
MappingConfig.ColumnItem rowKeyColumn = hbaseOrm.getRowKeyColumn();
int index = 0;
int i = 1;
boolean complete = false;
List<HRow> rows = new ArrayList<>();
out: for (Map<String, Object> r : data) {
byte[] rowKeyBytes;
if (hbaseOrm.getRowKey() != null) {
String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
// 判断是否有复合主键修改
for (String updateColumn : old.get(index).keySet()) {
for (String rowKeyColumnName : rowKeyColumns) {
if (rowKeyColumnName.equalsIgnoreCase(updateColumn)) {
// 调用删除插入操作
deleteAndInsert(config, dml);
continue out;
}
}
}
String rowKeyVale = getRowKeys(rowKeyColumns, r);
rowKeyBytes = Bytes.toBytes(rowKeyVale);
} else if (rowKeyColumn == null) {
Map<String, Object> rowKey = data.get(0);
rowKeyBytes = typeConvert(null, hbaseOrm, rowKey.values().iterator().next());
} else {
rowKeyBytes = typeConvert(rowKeyColumn, hbaseOrm, r.get(rowKeyColumn.getColumn()));
}
if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
Map<String, MappingConfig.ColumnItem> columnItems = hbaseOrm.getColumnItems();
HRow hRow = new HRow(rowKeyBytes);
for (String updateColumn : old.get(index).keySet()) {
MappingConfig.ColumnItem columnItem = columnItems.get(updateColumn);
if (columnItem == null) {
String family = hbaseOrm.getFamily();
String qualifier = updateColumn;
if (hbaseOrm.isUppercaseQualifier()) {
qualifier = qualifier.toUpperCase();
}
Object newVal = r.get(updateColumn);
if (newVal == null) {
hRow.addCell(family, qualifier, null);
} else {
hRow.addCell(family, qualifier, typeConvert(null, hbaseOrm, newVal));
}
} else {
// 排除修改id的情况
if (columnItem.isRowKey()) continue;
Object newVal = r.get(updateColumn);
if (newVal == null) {
hRow.addCell(columnItem.getFamily(), columnItem.getQualifier(), null);
} else {
hRow.addCell(columnItem.getFamily(),
columnItem.getQualifier(),
typeConvert(columnItem, hbaseOrm, newVal));
}
}
}
rows.add(hRow);
complete = false;
if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
rows.clear();
complete = true;
}
i++;
index++;
}
if (!complete && !rows.isEmpty()) {
hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
}
}
private void delete(MappingConfig config, Dml dml) {
List<Map<String, Object>> data = dml.getData();
if (data == null || data.isEmpty()) {
return;
}
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
// if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable());
// return;
// }
MappingConfig.ColumnItem rowKeyColumn = hbaseOrm.getRowKeyColumn();
boolean complete = false;
int i = 1;
Set<byte[]> rowKeys = new HashSet<>();
for (Map<String, Object> r : data) {
byte[] rowKeyBytes;
if (hbaseOrm.getRowKey() != null) {
String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
String rowKeyVale = getRowKeys(rowKeyColumns, r);
rowKeyBytes = Bytes.toBytes(rowKeyVale);
} else if (rowKeyColumn == null) {
// 如果不需要类型转换
Map<String, Object> rowKey = data.get(0);
rowKeyBytes = typeConvert(null, hbaseOrm, rowKey.values().iterator().next());
} else {
Object val = r.get(rowKeyColumn.getColumn());
rowKeyBytes = typeConvert(rowKeyColumn, hbaseOrm, val);
}
if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
rowKeys.add(rowKeyBytes);
complete = false;
if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rowKeys.isEmpty()) {
hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
rowKeys.clear();
complete = true;
}
i++;
}
if (!complete && !rowKeys.isEmpty()) {
hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
}
}
private void deleteAndInsert(MappingConfig config, Dml dml) {
List<Map<String, Object>> data = dml.getData();
List<Map<String, Object>> old = dml.getOld();
if (old == null || old.isEmpty() || data == null || data.isEmpty()) {
return;
}
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
int index = 0;
int i = 1;
boolean complete = false;
Set<byte[]> rowKeys = new HashSet<>();
List<HRow> rows = new ArrayList<>();
for (Map<String, Object> r : data) {
// 拼接老的rowKey
List<String> updateSubRowKey = new ArrayList<>();
for (String rowKeyColumnName : rowKeyColumns) {
for (String updateColumn : old.get(index).keySet()) {
if (rowKeyColumnName.equalsIgnoreCase(updateColumn)) {
updateSubRowKey.add(rowKeyColumnName);
}
}
}
if (updateSubRowKey.isEmpty()) {
throw new RuntimeException("没有更新复合主键的RowKey");
}
StringBuilder oldRowKey = new StringBuilder();
StringBuilder newRowKey = new StringBuilder();
for (String rowKeyColumnName : rowKeyColumns) {
newRowKey.append(r.get(rowKeyColumnName).toString()).append("|");
if (!updateSubRowKey.contains(rowKeyColumnName)) {
// 从data取
oldRowKey.append(r.get(rowKeyColumnName).toString()).append("|");
} else {
// 从old取
oldRowKey.append(old.get(index).get(rowKeyColumnName).toString()).append("|");
}
}
int len = newRowKey.length();
newRowKey.delete(len - 1, len);
len = oldRowKey.length();
oldRowKey.delete(len - 1, len);
byte[] newRowKeyBytes = Bytes.toBytes(newRowKey.toString());
byte[] oldRowKeyBytes = Bytes.toBytes(oldRowKey.toString());
rowKeys.add(oldRowKeyBytes);
HRow row = new HRow(newRowKeyBytes);
convertData2Row(hbaseOrm, row, r);
rows.add(row);
complete = false;
if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
rowKeys.clear();
rows.clear();
complete = true;
}
i++;
index++;
}
if (!complete && !rows.isEmpty()) {
hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
}
}
/**
* 根据对应的类型进行转换
*
* @param columnItem 列项配置
* @param hbaseOrm hbase映射配置
* @param value 值
* @return 复合字段rowKey
*/
private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm,
Object value) {
if (value == null) {
return null;
}
byte[] bytes = null;
if (columnItem == null || columnItem.getType() == null || "".equals(columnItem.getType())) {
if (MappingConfig.Mode.STRING == hbaseOrm.getMode()) {
bytes = Bytes.toBytes(value.toString());
} else if (MappingConfig.Mode.NATIVE == hbaseOrm.getMode()) {
bytes = TypeUtil.toBytes(value);
} else if (MappingConfig.Mode.PHOENIX == hbaseOrm.getMode()) {
PhType phType = PhType.getType(value.getClass());
bytes = PhTypeUtil.toBytes(value, phType);
}
} else {
if (hbaseOrm.getMode() == MappingConfig.Mode.STRING) {
bytes = Bytes.toBytes(value.toString());
} else if (hbaseOrm.getMode() == MappingConfig.Mode.NATIVE) {
Type type = Type.getType(columnItem.getType());
bytes = TypeUtil.toBytes(value, type);
} else if (hbaseOrm.getMode() == MappingConfig.Mode.PHOENIX) {
PhType phType = PhType.getType(columnItem.getType());
bytes = PhTypeUtil.toBytes(value, phType);
}
}
return bytes;
}
/**
* 获取复合字段作为rowKey的拼接
*
* @param rowKeyColumns 复合rowK对应的字段
* @param data 数据
* @return
*/
private static String getRowKeys(String[] rowKeyColumns, Map<String, Object> data) {
StringBuilder rowKeyValue = new StringBuilder();
for (String rowKeyColumnName : rowKeyColumns) {
Object obj = data.get(rowKeyColumnName);
if (obj != null) {
rowKeyValue.append(obj.toString());
}
rowKeyValue.append("|");
}
int len = rowKeyValue.length();
if (len > 0) {
rowKeyValue.delete(len - 1, len);
}
return rowKeyValue.toString();
}
}

View File

@@ -0,0 +1,84 @@
package com.alibaba.otter.canal.client.adapter.hbase.support;
import java.util.ArrayList;
import java.util.List;
/**
* HBase操作对象类
*
* @author machengyuan 2018-8-21 下午10:12:34
* @version 1.0.0
*/
public class HRow {
private byte[] rowKey;
private List<HCell> cells = new ArrayList<>();
public HRow(){
}
public HRow(byte[] rowKey){
this.rowKey = rowKey;
}
public byte[] getRowKey() {
return rowKey;
}
public void setRowKey(byte[] rowKey) {
this.rowKey = rowKey;
}
public List<HCell> getCells() {
return cells;
}
public void setCells(List<HCell> cells) {
this.cells = cells;
}
public void addCell(String family, String qualifier, byte[] value) {
HCell hCell = new HCell(family, qualifier, value);
cells.add(hCell);
}
public static class HCell {
private String family;
private String qualifier;
private byte[] value;
public HCell(){
}
public HCell(String family, String qualifier, byte[] value){
this.family = family;
this.qualifier = qualifier;
this.value = value;
}
public String getFamily() {
return family;
}
public void setFamily(String family) {
this.family = family;
}
public String getQualifier() {
return qualifier;
}
public void setQualifier(String qualifier) {
this.qualifier = qualifier;
}
public byte[] getValue() {
return value;
}
public void setValue(byte[] value) {
this.value = value;
}
}
}

View File

@@ -0,0 +1,159 @@
package com.alibaba.otter.canal.client.adapter.hbase.support;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* HBase操作模板
*
* @author machengyuan 2018-8-21 下午10:12:34
* @version 1.0.0
*/
public class HbaseTemplate {
private Logger logger = LoggerFactory.getLogger(this.getClass());
private Connection conn;
public HbaseTemplate(Connection conn){
this.conn = conn;
}
public boolean tableExists(String tableName) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
return admin.tableExists(TableName.valueOf(tableName));
} catch (IOException e) {
throw new RuntimeException(e);
}
}
public void createTable(String tableName, String... familyNames) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
// 添加列簇
if (familyNames != null) {
for (String familyName : familyNames) {
HColumnDescriptor hcd = new HColumnDescriptor(familyName);
desc.addFamily(hcd);
}
}
admin.createTable(desc);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
public void disableTable(String tableName) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
admin.disableTable(tableName);
} catch (IOException e) {
logger.error(e.getMessage(), e);
throw new RuntimeException(e);
}
}
public void deleteTable(String tableName) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
if (admin.isTableEnabled(tableName)) {
disableTable(tableName);
}
admin.deleteTable(tableName);
} catch (IOException e) {
logger.error(e.getMessage(), e);
throw new RuntimeException(e);
}
}
/**
* 插入一行数据
*
* @param tableName 表名
* @param hRow 行数据对象
* @return 是否成功
*/
public Boolean put(String tableName, HRow hRow) {
boolean flag = false;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
Put put = new Put(hRow.getRowKey());
for (HRow.HCell hCell : hRow.getCells()) {
put.addColumn(Bytes.toBytes(hCell.getFamily()), Bytes.toBytes(hCell.getQualifier()), hCell.getValue());
}
table.put(put);
flag = true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return flag;
}
/**
* 批量插入
*
* @param tableName 表名
* @param rows 行数据对象集合
* @return 是否成功
*/
public Boolean puts(String tableName, List<HRow> rows) {
boolean flag = false;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
List<Put> puts = new ArrayList<>();
for (HRow hRow : rows) {
Put put = new Put(hRow.getRowKey());
for (HRow.HCell hCell : hRow.getCells()) {
put.addColumn(Bytes.toBytes(hCell.getFamily()),
Bytes.toBytes(hCell.getQualifier()),
hCell.getValue());
}
puts.add(put);
}
if (!puts.isEmpty()) {
table.put(puts);
}
flag = true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return flag;
}
/**
* 批量删除数据
*
* @param tableName 表名
* @param rowKeys rowKey集合
* @return 是否成功
*/
public Boolean deletes(String tableName, Set<byte[]> rowKeys) {
boolean flag = false;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
List<Delete> deletes = new ArrayList<>();
for (byte[] rowKey : rowKeys) {
Delete delete = new Delete(rowKey);
deletes.add(delete);
}
if (!deletes.isEmpty()) {
table.delete(deletes);
}
flag = true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return flag;
}
}

View File

@@ -0,0 +1,150 @@
package com.alibaba.otter.canal.client.adapter.hbase.support;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.Date;
/**
* Phoenix类型
*
* @author machengyuan 2018-8-21 下午06:12:34
* @version 1.0.0
*/
public enum PhType {
DEFAULT(-1, "VARCHAR"),
UNSIGNED_INT(4, "UNSIGNED_INT"),
UNSIGNED_LONG(8, "UNSIGNED_LONG"),
UNSIGNED_TINYINT(1, "UNSIGNED_TINYINT"),
UNSIGNED_SMALLINT(2, "UNSIGNED_SMALLINT"),
UNSIGNED_FLOAT(4, "UNSIGNED_FLOAT"),
UNSIGNED_DOUBLE(8, "UNSIGNED_DOUBLE"),
INTEGER(4, "INTEGER"),
BIGINT(8, "BIGINT"),
TINYINT(1, "TINYINT"),
SMALLINT(2, "SMALLINT"),
FLOAT(4, "FLOAT"),
DOUBLE(8, "DOUBLE"),
DECIMAL(-1, "DECIMAL"),
BOOLEAN(1, "BOOLEAN"),
UNSIGNED_TIME(8, "UNSIGNED_TIME"),
UNSIGNED_DATE(8, "UNSIGNED_DATE"),
UNSIGNED_TIMESTAMP(12, "UNSIGNED_TIMESTAMP"),
TIME(8, "TIME"),
DATE(8, "DATE"),
TIMESTAMP(12, "TIMESTAMP"),
VARCHAR(-1, "VARCHAR"),
VARBINARY(-1, "VARBINARY");
/**
* -1:长度可变
*/
private int len;
private String type;
PhType(int len, String type) {
this.len = len;
this.type = type;
}
public int getLen() {
return len;
}
public String getType() {
return this.type;
}
public static PhType getType(Class<?> javaType) {
if (javaType == null) return DEFAULT;
PhType phType;
if (Integer.class.isAssignableFrom(javaType) || int.class.isAssignableFrom(javaType)) {
phType = INTEGER;
} else if (Long.class.isAssignableFrom(javaType) || long.class.isAssignableFrom(javaType)) {
phType = BIGINT;
} else if (Byte.class.isAssignableFrom(javaType) || byte.class.isAssignableFrom(javaType)) {
phType = TINYINT;
} else if (Short.class.isAssignableFrom(javaType) || short.class.isAssignableFrom(javaType)) {
phType = SMALLINT;
} else if (Float.class.isAssignableFrom(javaType) || float.class.isAssignableFrom(javaType)) {
phType = FLOAT;
} else if (Double.class.isAssignableFrom(javaType) || double.class.isAssignableFrom(javaType)) {
phType = DOUBLE;
} else if (Boolean.class.isAssignableFrom(javaType) || boolean.class.isAssignableFrom(javaType)) {
phType = BOOLEAN;
} else if (java.sql.Date.class.isAssignableFrom(javaType)) {
phType = DATE;
} else if (Time.class.isAssignableFrom(javaType)) {
phType = DATE;
} else if (Timestamp.class.isAssignableFrom(javaType)) {
phType = TIMESTAMP;
} else if (Date.class.isAssignableFrom(javaType)) {
phType = DATE;
} else if (byte[].class.isAssignableFrom(javaType)) {
phType = VARBINARY;
} else if (String.class.isAssignableFrom(javaType)) {
phType = VARCHAR;
} else if (BigDecimal.class.isAssignableFrom(javaType)) {
phType = DECIMAL;
} else if (BigInteger.class.isAssignableFrom(javaType)) {
phType = UNSIGNED_LONG;
} else {
phType = DEFAULT;
}
return phType;
}
public static PhType getType(String type) {
if (type == null) return DEFAULT;
PhType phType;
if (type.equalsIgnoreCase(UNSIGNED_INT.type)) {
phType = UNSIGNED_INT;
} else if (type.equalsIgnoreCase(UNSIGNED_LONG.type)) {
phType = UNSIGNED_LONG;
} else if (type.equalsIgnoreCase(UNSIGNED_TINYINT.type)) {
phType = UNSIGNED_TINYINT;
} else if (type.equalsIgnoreCase(UNSIGNED_SMALLINT.type)) {
phType = UNSIGNED_SMALLINT;
} else if (type.equalsIgnoreCase(UNSIGNED_FLOAT.type)) {
phType = UNSIGNED_FLOAT;
} else if (type.equalsIgnoreCase(UNSIGNED_DOUBLE.type)) {
phType = UNSIGNED_DOUBLE;
} else if (type.equalsIgnoreCase(INTEGER.type)) {
phType = INTEGER;
} else if (type.equalsIgnoreCase(BIGINT.type)) {
phType = BIGINT;
} else if (type.equalsIgnoreCase(TINYINT.type)) {
phType = TINYINT;
} else if (type.equalsIgnoreCase(SMALLINT.type)) {
phType = SMALLINT;
} else if (type.equalsIgnoreCase(FLOAT.type)) {
phType = FLOAT;
} else if (type.equalsIgnoreCase(DOUBLE.type)) {
phType = DOUBLE;
} else if (type.equalsIgnoreCase(BOOLEAN.type)) {
phType = BOOLEAN;
} else if (type.equalsIgnoreCase(UNSIGNED_TIME.type)) {
phType = UNSIGNED_TIME;
} else if (type.equalsIgnoreCase(UNSIGNED_DATE.type)) {
phType = UNSIGNED_DATE;
} else if (type.equalsIgnoreCase(UNSIGNED_TIMESTAMP.type)) {
phType = UNSIGNED_TIMESTAMP;
} else if (type.equalsIgnoreCase(TIME.type)) {
phType = TIME;
} else if (type.equalsIgnoreCase(DATE.type)) {
phType = DATE;
} else if (type.equalsIgnoreCase(TIMESTAMP.type)) {
phType = TIMESTAMP;
} else if (type.equalsIgnoreCase(VARCHAR.type)) {
phType = VARCHAR;
} else if (type.equalsIgnoreCase(VARBINARY.type)) {
phType = VARBINARY;
} else if (type.equalsIgnoreCase(DECIMAL.type)) {
phType = DECIMAL;
} else {
phType = DEFAULT;
}
return phType;
}
}

View File

@@ -0,0 +1,614 @@
package com.alibaba.otter.canal.client.adapter.hbase.support;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.math.MathContext;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.Date;
import org.apache.hadoop.hbase.util.Bytes;
import org.joda.time.DateTime;
import com.google.common.math.LongMath;
/**
* Phoenix类型转换工具类
*
* @author machengyuan 2018-8-21 下午06:14:26
* @version 1.0.0
*/
public class PhTypeUtil {
public static byte[] toBytes(Object v, PhType phType) {
if (v == null) return null;
byte[] b = null;
if (phType == PhType.DEFAULT) {
PhType phType1 = PhType.getType(v.getClass());
if (phType1 != null && phType1 != PhType.DEFAULT) {
toBytes(v, phType1);
}
} else if (phType == PhType.INTEGER) {
b = new byte[Bytes.SIZEOF_INT];
encodeInt(((Number) v).intValue(), b, 0);
} else if (phType == PhType.UNSIGNED_INT) {
b = new byte[Bytes.SIZEOF_INT];
encodeUnsignedInt(((Number) v).intValue(), b, 0);
} else if (phType == PhType.BIGINT) {
b = new byte[Bytes.SIZEOF_LONG];
encodeLong(((Number) v).longValue(), b, 0);
} else if (phType == PhType.UNSIGNED_LONG) {
b = new byte[Bytes.SIZEOF_LONG];
encodeUnsignedLong(((Number) v).longValue(), b, 0);
} else if (phType == PhType.SMALLINT) {
b = new byte[Bytes.SIZEOF_SHORT];
encodeShort(((Number) v).shortValue(), b, 0);
} else if (phType == PhType.UNSIGNED_SMALLINT) {
b = new byte[Bytes.SIZEOF_SHORT];
encodeUnsignedShort(((Number) v).shortValue(), b, 0);
} else if (phType == PhType.TINYINT) {
b = new byte[Bytes.SIZEOF_BYTE];
encodeByte(((Number) v).byteValue(), b, 0);
} else if (phType == PhType.UNSIGNED_TINYINT) {
b = new byte[Bytes.SIZEOF_BYTE];
encodeUnsignedByte(((Number) v).byteValue(), b, 0);
} else if (phType == PhType.FLOAT) {
b = new byte[Bytes.SIZEOF_FLOAT];
encodeFloat(((Number) v).floatValue(), b, 0);
} else if (phType == PhType.UNSIGNED_FLOAT) {
b = new byte[Bytes.SIZEOF_FLOAT];
encodeUnsignedFloat(((Number) v).floatValue(), b, 0);
} else if (phType == PhType.DOUBLE) {
b = new byte[Bytes.SIZEOF_DOUBLE];
encodeDouble(((Number) v).doubleValue(), b, 0);
} else if (phType == PhType.UNSIGNED_DOUBLE) {
b = new byte[Bytes.SIZEOF_DOUBLE];
encodeUnsignedDouble(((Number) v).doubleValue(), b, 0);
} else if (phType == PhType.BOOLEAN) {
if ((Boolean) v) {
b = new byte[] { 1 };
} else {
b = new byte[] { 0 };
}
} else if (phType == PhType.TIME || phType == PhType.DATE) {
b = new byte[Bytes.SIZEOF_LONG];
encodeDate(v, b, 0);
} else if (phType == PhType.TIMESTAMP) {
b = new byte[Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT];
encodeTimestamp(v, b, 0);
} else if (phType == PhType.UNSIGNED_TIME || phType == PhType.UNSIGNED_DATE) {
b = new byte[Bytes.SIZEOF_LONG];
encodeUnsignedDate(v, b, 0);
} else if (phType == PhType.UNSIGNED_TIMESTAMP) {
b = new byte[Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT];
encodeUnsignedTimestamp(v, b, 0);
} else if (phType == PhType.VARBINARY) {
b = (byte[]) v;
} else if (phType == PhType.VARCHAR) {
b = Bytes.toBytes(v.toString());
} else if (phType == PhType.DECIMAL) {
if (v instanceof BigDecimal) {
b = encodeDecimal(v);
} else if (v instanceof Number) {
b = encodeDecimal(new BigDecimal(v.toString()));
}
}
return b;
}
public static Object toObject(byte[] b, PhType phType) {
if (b == null) return null;
Object v = null;
if (phType == PhType.INTEGER) {
v = decodeInt(b, 0);
} else if (phType == PhType.UNSIGNED_INT) {
v = decodeUnsignedInt(b, 0);
} else if (phType == PhType.BIGINT) {
v = decodeLong(b, 0);
} else if (phType == PhType.UNSIGNED_LONG) {
v = decodeUnsignedLong(b, 0);
} else if (phType == PhType.SMALLINT) {
v = decodeShort(b, 0);
} else if (phType == PhType.UNSIGNED_SMALLINT) {
v = decodeUnsignedShort(b, 0);
} else if (phType == PhType.TINYINT) {
v = decodeByte(b, 0);
} else if (phType == PhType.UNSIGNED_TINYINT) {
v = decodeUnsignedByte(b, 0);
} else if (phType == PhType.FLOAT) {
v = decodeFloat(b, 0);
} else if (phType == PhType.UNSIGNED_FLOAT) {
v = decodeUnsignedFloat(b, 0);
} else if (phType == PhType.DOUBLE) {
v = decodeDouble(b, 0);
} else if (phType == PhType.UNSIGNED_DOUBLE) {
v = decodeUnsignedDouble(b, 0);
} else if (phType == PhType.BOOLEAN) {
checkForSufficientLength(b, 0, Bytes.SIZEOF_BOOLEAN);
if (b[0] == 1) {
v = true;
} else if (b[0] == 0) {
v = false;
}
} else if (phType == PhType.TIME || phType == PhType.DATE) {
v = new Date(decodeLong(b, 0));
} else if (phType == PhType.TIMESTAMP) {
long millisDeserialized = decodeLong(b, 0);
Timestamp ts = new Timestamp(millisDeserialized);
int nanosDeserialized = decodeUnsignedInt(b, Bytes.SIZEOF_LONG);
ts.setNanos(nanosDeserialized < 1000000 ? ts.getNanos() + nanosDeserialized : nanosDeserialized);
v = ts;
} else if (phType == PhType.UNSIGNED_TIME || phType == PhType.UNSIGNED_DATE) {
v = new Date(decodeUnsignedLong(b, 0));
} else if (phType == PhType.UNSIGNED_TIMESTAMP) {
long millisDeserialized = decodeUnsignedLong(b, 0);
Timestamp ts = new Timestamp(millisDeserialized);
int nanosDeserialized = decodeUnsignedInt(b, Bytes.SIZEOF_LONG);
ts.setNanos(nanosDeserialized < 1000000 ? ts.getNanos() + nanosDeserialized : nanosDeserialized);
v = ts;
} else if (phType == PhType.VARBINARY) {
v = b;
} else if (phType == PhType.VARCHAR || phType == PhType.DEFAULT) {
v = Bytes.toString(b);
} else if (phType == PhType.DECIMAL) {
v = decodeDecimal(b, 0, b.length);
}
return v;
}
private static int decodeInt(byte[] bytes, int o) {
checkForSufficientLength(bytes, o, Bytes.SIZEOF_INT);
int v;
v = bytes[o] ^ 0x80; // Flip sign bit back
for (int i = 1; i < Bytes.SIZEOF_INT; i++) {
v = (v << 8) + (bytes[o + i] & 0xff);
}
return v;
}
private static int encodeInt(int v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
b[o + 0] = (byte) ((v >> 24) ^ 0x80); // Flip sign bit so that INTEGER is binary comparable
b[o + 1] = (byte) (v >> 16);
b[o + 2] = (byte) (v >> 8);
b[o + 3] = (byte) v;
return Bytes.SIZEOF_INT;
}
private static int decodeUnsignedInt(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
int v = Bytes.toInt(b, o);
if (v < 0) {
throw new RuntimeException();
}
return v;
}
private static int encodeUnsignedInt(int v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
if (v < 0) {
throw new RuntimeException();
}
Bytes.putInt(b, o, v);
return Bytes.SIZEOF_INT;
}
private static long decodeLong(byte[] bytes, int o) {
checkForSufficientLength(bytes, o, Bytes.SIZEOF_LONG);
long v;
byte b = bytes[o];
v = b ^ 0x80; // Flip sign bit back
for (int i = 1; i < Bytes.SIZEOF_LONG; i++) {
b = bytes[o + i];
v = (v << 8) + (b & 0xff);
}
return v;
}
private static int encodeLong(long v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
b[o + 0] = (byte) ((v >> 56) ^ 0x80); // Flip sign bit so that INTEGER is binary comparable
b[o + 1] = (byte) (v >> 48);
b[o + 2] = (byte) (v >> 40);
b[o + 3] = (byte) (v >> 32);
b[o + 4] = (byte) (v >> 24);
b[o + 5] = (byte) (v >> 16);
b[o + 6] = (byte) (v >> 8);
b[o + 7] = (byte) v;
return Bytes.SIZEOF_LONG;
}
private static long decodeUnsignedLong(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
long v = 0;
for (int i = o; i < o + Bytes.SIZEOF_LONG; i++) {
v <<= 8;
v ^= b[i] & 0xFF;
}
if (v < 0) {
throw new RuntimeException();
}
return v;
}
private static int encodeUnsignedLong(long v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
if (v < 0) {
throw new RuntimeException();
}
Bytes.putLong(b, o, v);
return Bytes.SIZEOF_LONG;
}
private static short decodeShort(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
int v;
v = b[o] ^ 0x80; // Flip sign bit back
for (int i = 1; i < Bytes.SIZEOF_SHORT; i++) {
v = (v << 8) + (b[o + i] & 0xff);
}
return (short) v;
}
private static int encodeShort(short v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
b[o + 0] = (byte) ((v >> 8) ^ 0x80); // Flip sign bit so that Short is binary comparable
b[o + 1] = (byte) v;
return Bytes.SIZEOF_SHORT;
}
private static short decodeUnsignedShort(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
short v = Bytes.toShort(b, o);
if (v < 0) {
throw new RuntimeException();
}
return v;
}
private static int encodeUnsignedShort(short v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
if (v < 0) {
throw new RuntimeException();
}
Bytes.putShort(b, o, v);
return Bytes.SIZEOF_SHORT;
}
private static byte decodeByte(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
int v;
v = b[o] ^ 0x80; // Flip sign bit back
return (byte) v;
}
private static int encodeByte(byte v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
b[o] = (byte) (v ^ 0x80); // Flip sign bit so that Short is binary comparable
return Bytes.SIZEOF_BYTE;
}
private static byte decodeUnsignedByte(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
byte v = b[o];
if (v < 0) {
throw new RuntimeException();
}
return v;
}
private static int encodeUnsignedByte(byte v, byte[] b, int o) {
if (v < 0) {
throw new RuntimeException();
}
Bytes.putByte(b, o, v);
return Bytes.SIZEOF_BYTE;
}
private static float decodeFloat(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
int value;
value = Bytes.toInt(b, o);
value--;
value ^= (~value >> Integer.SIZE - 1) | Integer.MIN_VALUE;
return Float.intBitsToFloat(value);
}
private static int encodeFloat(float v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_FLOAT);
int i = Float.floatToIntBits(v);
i = (i ^ ((i >> Integer.SIZE - 1) | Integer.MIN_VALUE)) + 1;
Bytes.putInt(b, o, i);
return Bytes.SIZEOF_FLOAT;
}
private static float decodeUnsignedFloat(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_FLOAT);
float v = Bytes.toFloat(b, o);
if (v < 0) {
throw new RuntimeException();
}
return v;
}
private static int encodeUnsignedFloat(float v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_FLOAT);
if (v < 0) {
throw new RuntimeException();
}
Bytes.putFloat(b, o, v);
return Bytes.SIZEOF_FLOAT;
}
private static double decodeDouble(byte[] bytes, int o) {
checkForSufficientLength(bytes, o, Bytes.SIZEOF_LONG);
long l;
l = Bytes.toLong(bytes, o);
l--;
l ^= (~l >> Long.SIZE - 1) | Long.MIN_VALUE;
return Double.longBitsToDouble(l);
}
private static int encodeDouble(double v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
long l = Double.doubleToLongBits(v);
l = (l ^ ((l >> Long.SIZE - 1) | Long.MIN_VALUE)) + 1;
Bytes.putLong(b, o, l);
return Bytes.SIZEOF_LONG;
}
private static double decodeUnsignedDouble(byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_DOUBLE);
double v = Bytes.toDouble(b, o);
if (v < 0) {
throw new RuntimeException();
}
return v;
}
private static int encodeUnsignedDouble(double v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_DOUBLE);
if (v < 0) {
throw new RuntimeException();
}
Bytes.putDouble(b, o, v);
return Bytes.SIZEOF_DOUBLE;
}
private static int encodeDate(Object v, byte[] b, int o) {
if (v instanceof Date) {
encodeLong(((Date) v).getTime(), b, 0);
} else if (v instanceof String) {
String dateStr = (String) v;
Date date;
try {
date = parseDatetime(dateStr);
if (date != null) {
encodeLong(date.getTime(), b, 0);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
return Bytes.SIZEOF_LONG;
}
private static int encodeTimestamp(Object v, byte[] b, int o) {
if (v instanceof Timestamp) {
Timestamp ts = (Timestamp) v;
encodeLong(ts.getTime(), b, o);
Bytes.putInt(b, Bytes.SIZEOF_LONG, ts.getNanos() % 1000000);
} else {
encodeDate(v, b, o);
}
return Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
}
private static int encodeUnsignedDate(Object v, byte[] b, int o) {
if (v instanceof Date) {
encodeUnsignedLong(((Date) v).getTime(), b, 0);
} else if (v instanceof String) {
String dateStr = (String) v;
Date date;
try {
date = parseDatetime(dateStr);
if (date != null) {
encodeUnsignedLong(date.getTime(), b, 0);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
return Bytes.SIZEOF_LONG;
}
private static int encodeUnsignedTimestamp(Object v, byte[] b, int o) {
if (v instanceof Timestamp) {
Timestamp ts = (Timestamp) v;
encodeUnsignedLong(ts.getTime(), b, o);
Bytes.putInt(b, Bytes.SIZEOF_LONG, ts.getNanos() % 1000000);
} else {
encodeUnsignedDate(v, b, o);
}
return Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
}
private static byte[] encodeDecimal(Object object) {
if (object == null) {
return new byte[0];
}
BigDecimal v = (BigDecimal) object;
v = v.round(DEFAULT_MATH_CONTEXT).stripTrailingZeros();
int len = getLength(v);
byte[] result = new byte[Math.min(len, 21)];
decimalToBytes(v, result, 0, len);
return result;
}
private static BigDecimal decodeDecimal(byte[] bytes, int offset, int length) {
if (length == 1 && bytes[offset] == ZERO_BYTE) {
return BigDecimal.ZERO;
}
int signum = ((bytes[offset] & 0x80) == 0) ? -1 : 1;
int scale;
int index;
int digitOffset;
long multiplier = 100L;
int begIndex = offset + 1;
if (signum == 1) {
scale = (byte) (((bytes[offset] & 0x7F) - 65) * -2);
index = offset + length;
digitOffset = POS_DIGIT_OFFSET;
} else {
scale = (byte) ((~bytes[offset] - 65 - 128) * -2);
index = offset + length - (bytes[offset + length - 1] == NEG_TERMINAL_BYTE ? 1 : 0);
digitOffset = -NEG_DIGIT_OFFSET;
}
length = index - offset;
long l = signum * bytes[--index] - digitOffset;
if (l % 10 == 0) { // trailing zero
scale--; // drop trailing zero and compensate in the scale
l /= 10;
multiplier = 10;
}
// Use long arithmetic for as long as we can
while (index > begIndex) {
if (l >= MAX_LONG_FOR_DESERIALIZE || multiplier >= Long.MAX_VALUE / 100) {
multiplier = LongMath.divide(multiplier, 100L, RoundingMode.UNNECESSARY);
break; // Exit loop early so we don't overflow our multiplier
}
int digit100 = signum * bytes[--index] - digitOffset;
l += digit100 * multiplier;
multiplier = LongMath.checkedMultiply(multiplier, 100);
}
BigInteger bi;
// If still more digits, switch to BigInteger arithmetic
if (index > begIndex) {
bi = BigInteger.valueOf(l);
BigInteger biMultiplier = BigInteger.valueOf(multiplier).multiply(ONE_HUNDRED);
do {
int digit100 = signum * bytes[--index] - digitOffset;
bi = bi.add(biMultiplier.multiply(BigInteger.valueOf(digit100)));
biMultiplier = biMultiplier.multiply(ONE_HUNDRED);
} while (index > begIndex);
if (signum == -1) {
bi = bi.negate();
}
} else {
bi = BigInteger.valueOf(l * signum);
}
// Update the scale based on the precision
scale += (length - 2) * 2;
BigDecimal v = new BigDecimal(bi, scale);
return v;
}
private static int getLength(BigDecimal v) {
int signum = v.signum();
if (signum == 0) { // Special case for zero
return 1;
}
return (signum < 0 ? 2 : 1) + (v.precision() + 1 + (v.scale() % 2 == 0 ? 0 : 1)) / 2;
}
private static final int MAX_PRECISION = 38;
private static final MathContext DEFAULT_MATH_CONTEXT = new MathContext(MAX_PRECISION, RoundingMode.HALF_UP);
private static final Integer MAX_BIG_DECIMAL_BYTES = 21;
private static final byte ZERO_BYTE = (byte) 0x80;
private static final byte NEG_TERMINAL_BYTE = (byte) 102;
private static final int EXP_BYTE_OFFSET = 65;
private static final int POS_DIGIT_OFFSET = 1;
private static final int NEG_DIGIT_OFFSET = 101;
private static final BigInteger MAX_LONG = BigInteger.valueOf(Long.MAX_VALUE);
private static final BigInteger MIN_LONG = BigInteger.valueOf(Long.MIN_VALUE);
private static final BigInteger ONE_HUNDRED = BigInteger.valueOf(100);
private static final long MAX_LONG_FOR_DESERIALIZE = Long.MAX_VALUE / 1000;
private static int decimalToBytes(BigDecimal v, byte[] result, final int offset, int length) {
int signum = v.signum();
if (signum == 0) {
result[offset] = ZERO_BYTE;
return 1;
}
int index = offset + length;
int scale = v.scale();
int expOffset = scale % 2 * (scale < 0 ? -1 : 1);
int multiplyBy;
BigInteger divideBy;
if (expOffset == 0) {
multiplyBy = 1;
divideBy = ONE_HUNDRED;
} else {
multiplyBy = 10;
divideBy = BigInteger.TEN;
}
// Normalize the scale based on what is necessary to end up with a base 100
// decimal (i.e. 10.123e3)
int digitOffset;
BigInteger compareAgainst;
if (signum == 1) {
digitOffset = POS_DIGIT_OFFSET;
compareAgainst = MAX_LONG;
scale -= (length - 2) * 2;
result[offset] = (byte) ((-(scale + expOffset) / 2 + EXP_BYTE_OFFSET) | 0x80);
} else {
digitOffset = NEG_DIGIT_OFFSET;
compareAgainst = MIN_LONG;
// Scale adjustment shouldn't include terminal byte in length
scale -= (length - 2 - 1) * 2;
result[offset] = (byte) (~(-(scale + expOffset) / 2 + EXP_BYTE_OFFSET + 128) & 0x7F);
if (length <= MAX_BIG_DECIMAL_BYTES) {
result[--index] = NEG_TERMINAL_BYTE;
} else {
// Adjust length and offset down because we don't have enough room
length = MAX_BIG_DECIMAL_BYTES;
index = offset + length;
}
}
BigInteger bi = v.unscaledValue();
// Use BigDecimal arithmetic until we can fit into a long
while (bi.compareTo(compareAgainst) * signum > 0) {
BigInteger[] dandr = bi.divideAndRemainder(divideBy);
bi = dandr[0];
int digit = dandr[1].intValue();
result[--index] = (byte) (digit * multiplyBy + digitOffset);
multiplyBy = 1;
divideBy = ONE_HUNDRED;
}
long l = bi.longValue();
do {
long divBy = 100 / multiplyBy;
long digit = l % divBy;
l /= divBy;
result[--index] = (byte) (digit * multiplyBy + digitOffset);
multiplyBy = 1;
} while (l != 0);
return length;
}
private static void checkForSufficientLength(byte[] b, int offset, int requiredLength) {
if (b.length < offset + requiredLength) {
throw new RuntimeException(
"Expected length of at least " + requiredLength + " bytes, but had " + (b.length - offset));
}
}
private static Date parseDatetime(String dateStr) {
Date date = null;
int len = dateStr.length();
if (len == 10 && dateStr.charAt(4) == '-' && dateStr.charAt(7) == '-') {
date = new DateTime(dateStr).toDate();
} else if (len == 8 && dateStr.charAt(2) == ':' && dateStr.charAt(5) == ':') {
date = new DateTime("T" + dateStr).toDate();
} else if (len >= 19 && dateStr.charAt(4) == '-' && dateStr.charAt(7) == '-' && dateStr.charAt(13) == ':'
&& dateStr.charAt(16) == ':') {
date = new DateTime(dateStr.replace(" ", "T")).toDate();
}
return date;
}
}

View File

@@ -0,0 +1,104 @@
package com.alibaba.otter.canal.client.adapter.hbase.support;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.Date;
/**
* Java类型
*
* @author machengyuan 2018-8-21 下午06:11:36
* @version 1.0.0
*/
public enum Type {
DEFAULT("STRING"),
STRING("STRING"),
INTEGER("INTEGER"),
LONG("LONG"),
SHORT("SHORT"),
BOOLEAN("BOOLEAN"),
FLOAT("FLOAT"),
DOUBLE("DOUBLE"),
BIGDECIMAL("BIGDECIMAL"),
DATE("DATE"),
BYTE("BYTE"),
BYTES("BYTES");
private String type;
Type(String type) {
this.type = type;
}
public String getType() {
return type;
}
public static Type getType(String type) {
if (type == null) {
return DEFAULT;
}
Type res;
if (type.equalsIgnoreCase("STRING")) {
res = STRING;
} else if (type.equalsIgnoreCase("INTEGER")) {
res = INTEGER;
} else if (type.equalsIgnoreCase("LONG")) {
res = LONG;
} else if (type.equalsIgnoreCase("SHORT")) {
res = SHORT;
} else if (type.equalsIgnoreCase("BOOLEAN")) {
res = BOOLEAN;
} else if (type.equalsIgnoreCase("FLOAT")) {
res = FLOAT;
} else if (type.equalsIgnoreCase("DOUBLE")) {
res = DOUBLE;
} else if (type.equalsIgnoreCase("BIGDECIMAL")) {
res = BIGDECIMAL;
} else if (type.equalsIgnoreCase("DATE")) {
res = DATE;
} else if (type.equalsIgnoreCase("BYTE")) {
res = BYTE;
} else if (type.equalsIgnoreCase("BYTES")) {
res = BYTES;
} else {
res = DEFAULT;
}
return res;
}
public static Type getType(Class<?> javaType) {
if (javaType == null) {
return DEFAULT;
}
Type type;
if (Integer.class.isAssignableFrom(javaType) || int.class.isAssignableFrom(javaType)) {
type = INTEGER;
} else if (Long.class.isAssignableFrom(javaType) || long.class.isAssignableFrom(javaType)) {
type = LONG;
} else if (Byte.class.isAssignableFrom(javaType) || byte.class.isAssignableFrom(javaType)) {
type = BYTE;
} else if (Short.class.isAssignableFrom(javaType) || short.class.isAssignableFrom(javaType)) {
type = SHORT;
} else if (Float.class.isAssignableFrom(javaType) || float.class.isAssignableFrom(javaType)) {
type = FLOAT;
} else if (Double.class.isAssignableFrom(javaType) || double.class.isAssignableFrom(javaType)) {
type = DOUBLE;
} else if (Boolean.class.isAssignableFrom(javaType) || boolean.class.isAssignableFrom(javaType)) {
type = BOOLEAN;
} else if (Date.class.isAssignableFrom(javaType)) {
type = DATE;
} else if (byte[].class.isAssignableFrom(javaType)) {
type = BYTES;
} else if (String.class.isAssignableFrom(javaType)) {
type = STRING;
} else if (BigDecimal.class.isAssignableFrom(javaType)) {
type = BIGDECIMAL;
} else if (BigInteger.class.isAssignableFrom(javaType)) {
type = LONG;
} else {
type = DEFAULT;
}
return type;
}
}

View File

@@ -0,0 +1,188 @@
package com.alibaba.otter.canal.client.adapter.hbase.support;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.Date;
import org.apache.hadoop.hbase.util.Bytes;
/**
* Java类型转换工具类
*
* @author machengyuan 2018-8-21 下午06:12:34
* @version 1.0.0
*/
public class TypeUtil {
public static byte[] toBytes(Object obj) {
if (obj == null) {
return null;
}
byte[] bytes;
if (obj instanceof String) {
bytes = Bytes.toBytes((String) obj);
} else if (obj instanceof Integer) {
bytes = Bytes.toBytes((Integer) obj);
} else if (obj instanceof Long) {
bytes = Bytes.toBytes((Long) obj);
} else if (obj instanceof Short) {
bytes = Bytes.toBytes((Short) obj);
} else if (obj instanceof Boolean) {
bytes = Bytes.toBytes((Boolean) obj);
} else if (obj instanceof Float) {
bytes = Bytes.toBytes((Float) obj);
} else if (obj instanceof Double) {
bytes = Bytes.toBytes((Double) obj);
} else if (obj instanceof BigDecimal) {
bytes = Bytes.toBytes((BigDecimal) obj);
} else if (obj instanceof BigInteger) {
bytes = Bytes.toBytes(((BigInteger) obj).longValue());
} else if (obj instanceof Date) {
bytes = Bytes.toBytes(((Date) obj).getTime());
} else if (obj instanceof Byte) {
bytes = new byte[] { (byte) obj };
} else if (obj instanceof byte[]) {
bytes = (byte[]) obj;
} else {
// 其余类型统一转换为string
bytes = Bytes.toBytes(obj.toString());
}
return bytes;
}
public static byte[] toBytes(Object v, Type type) {
if (v == null) {
return null;
}
byte[] b = null;
if (type == Type.DEFAULT) {
Type type1 = Type.getType(v.getClass());
if (type1 != null && type1 != Type.DEFAULT) {
b = toBytes(v, type1);
}
} else if (type == Type.STRING) {
b = Bytes.toBytes(v.toString());
} else if (type == Type.INTEGER) {
b = Bytes.toBytes(((Number) v).intValue());
} else if (type == Type.LONG) {
b = Bytes.toBytes(((Number) v).longValue());
} else if (type == Type.SHORT) {
b = Bytes.toBytes(((Number) v).shortValue());
} else if (type == Type.BYTE) {
b = Bytes.toBytes(((Number) v).byteValue());
} else if (type == Type.FLOAT) {
b = Bytes.toBytes(((Number) v).floatValue());
} else if (type == Type.DOUBLE) {
b = Bytes.toBytes(((Number) v).doubleValue());
} else if (type == Type.BOOLEAN) {
b = Bytes.toBytes(((Boolean) v));
} else if (type == Type.DATE) {
b = Bytes.toBytes(((Date) v).getTime());
} else if (type == Type.BYTES) {
b = (byte[]) v;
} else if (type == Type.BIGDECIMAL) {
if (v instanceof BigDecimal) {
b = Bytes.toBytes((BigDecimal) v);
} else {
b = Bytes.toBytes(new BigDecimal(v.toString()));
}
}
return b;
}
public static <T> T toObject(byte[] bytes, Class<T> clazz) {
if (bytes == null) {
return null;
}
Object res;
if (String.class.isAssignableFrom(clazz)) {
res = Bytes.toString(bytes);
} else if (Integer.class.isAssignableFrom(clazz)) {
res = Bytes.toInt(bytes);
} else if (Long.class.isAssignableFrom(clazz)) {
res = Bytes.toLong(bytes);
} else if (Short.class.isAssignableFrom(clazz)) {
res = Bytes.toShort(bytes);
} else if (Boolean.class.isAssignableFrom(clazz)) {
res = Bytes.toBoolean(bytes);
} else if (Float.class.isAssignableFrom(clazz)) {
res = Bytes.toFloat(bytes);
} else if (Double.class.isAssignableFrom(clazz)) {
res = Bytes.toDouble(bytes);
} else if (BigDecimal.class.isAssignableFrom(clazz)) {
res = Bytes.toBigDecimal(bytes);
} else if (BigInteger.class.isAssignableFrom(clazz)) {
res = Bytes.toLong(bytes);
} else if (java.sql.Date.class.isAssignableFrom(clazz)) {
long ts = Bytes.toLong(bytes);
res = new java.sql.Date(ts);
} else if (Time.class.isAssignableFrom(clazz)) {
long ts = Bytes.toLong(bytes);
res = new Time(ts);
} else if (Timestamp.class.isAssignableFrom(clazz)) {
long ts = Bytes.toLong(bytes);
res = new Timestamp(ts);
} else if (Date.class.isAssignableFrom(clazz)) {
long ts = Bytes.toLong(bytes);
res = new Date(ts);
} else if (Byte.class.isAssignableFrom(clazz)) {
res = bytes[0];
} else {
throw new IllegalArgumentException("mismatch class type");
}
// noinspection unchecked
return (T) res;
}
public static <T> T toObject(byte[] bytes, Type type) {
if (bytes == null) {
return null;
}
Object res = null;
if (type == Type.STRING || type == Type.DEFAULT) {
res = Bytes.toString(bytes);
} else if (type == Type.INTEGER) {
if (bytes.length == Bytes.SIZEOF_INT) {
res = Bytes.toInt(bytes);
}
} else if (type == Type.LONG) {
if (bytes.length == Bytes.SIZEOF_LONG) {
res = Bytes.toLong(bytes);
}
} else if (type == Type.SHORT) {
if (bytes.length == Bytes.SIZEOF_SHORT) {
res = Bytes.toShort(bytes);
}
} else if (type == Type.BYTE) {
if (bytes.length == Bytes.SIZEOF_BYTE) {
res = bytes[0];
}
} else if (type == Type.FLOAT) {
if (bytes.length == Bytes.SIZEOF_FLOAT) {
res = Bytes.toFloat(bytes);
}
} else if (type == Type.DOUBLE) {
if (bytes.length == Bytes.SIZEOF_DOUBLE) {
res = Bytes.toDouble(bytes);
}
} else if (type == Type.BOOLEAN) {
if (bytes.length == Bytes.SIZEOF_BOOLEAN) {
res = Bytes.toBoolean(bytes);
}
} else if (type == Type.DATE) {
if (bytes.length == Bytes.SIZEOF_LONG) {
res = new Date(Bytes.toLong(bytes));
}
} else if (type == Type.BYTES) {
res = bytes;
} else if (type == Type.BIGDECIMAL) {
res = Bytes.toBigDecimal(bytes);
} else {
throw new IllegalArgumentException("mismatch class type");
}
// noinspection unchecked
return (T) res;
}
}

View File

@@ -0,0 +1 @@
hbase=com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter

View File

@@ -0,0 +1,5 @@
# 详细映射配置
mytest_person2.yml
# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认
mytest.person

View File

@@ -0,0 +1,56 @@
hbaseOrm:
mode: PHOENIX #NATIVE #STRING
database: mytest # 数据库名
table: person2 # 数据库表名
hbaseTable: MYTEST.PERSON2 # HBase表名
family: CF # 默认统一Family名称
uppercaseQualifier: true # 字段名转大写, 默认为true
commitBatch: 3000 # 批量提交的大小
#rowKey: id,type # 复合字段rowKey不能和columns中的rowKey重复
columns:
# 数据库字段:HBase对应字段
id: ROWKEY$UNSIGNED_LONG
name: NAME
email: EMAIL
type: $DECIMAL
c_time: C_TIME$UNSIGNED_TIMESTAMP
birthday: BIRTHDAY$DATE
# -- NATIVE类型
# $DEFAULT
# $STRING
# $INTEGER
# $LONG
# $SHORT
# $BOOLEAN
# $FLOAT
# $DOUBLE
# $BIGDECIMAL
# $DATE
# $BYTE
# $BYTES
# -- PHOENIX类型
# $DEFAULT 对应PHOENIX里的VARCHAR
# $UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节
# $UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节
# $UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节
# $UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节
# $UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节
# $UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节
# $INTEGER 对应PHOENIX里的INTEGER 4字节
# $BIGINT 对应PHOENIX里的BIGINT 8字节
# $TINYINT 对应PHOENIX里的TINYINT 1字节
# $SMALLINT 对应PHOENIX里的SMALLINT 2字节
# $FLOAT 对应PHOENIX里的FLOAT 4字节
# DOUBLE 对应PHOENIX里的DOUBLE 8字节
# $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节
# $TIME 对应PHOENIX里的TIME 8字节
# $DATE 对应PHOENIX里的DATE 8字节
# $TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节
# $UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节
# $UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节
# $UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节
# $VARCHAR 对应PHOENIX里的VARCHAR 动态长度
# $VARBINARY 对应PHOENIX里的VARBINARY 动态长度
# $DECIMAL 对应PHOENIX里的DECIMAL 动态长度

View File

@@ -0,0 +1,47 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.logger</artifactId>
<packaging>jar</packaging>
<name>canal client adapter logger example module for otter ${project.version}</name>
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

View File

@@ -0,0 +1,45 @@
package com.alibaba.otter.canal.client.adapter.logger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.protocol.Message;
/**
* 外部适配器示例
*
* @author machengyuan 2018-8-19 下午11:45:38
* @version 1.0.0
*/
@SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name
public class LoggerAdapterExample implements CanalOuterAdapter {
private Logger logger = LoggerFactory.getLogger(this.getClass());
@Override
public void writeOut(Message message) {
// 直接输出日志信息
MessageUtil.parse4Dml(message, new MessageUtil.Consumer<Dml>() {
@Override
public void accept(Dml dml) {
logger.info(dml.toString());
}
});
}
@Override
public void init(CanalOuterAdapterConfiguration configuration) {
}
@Override
public void destroy() {
}
}

View File

@@ -0,0 +1 @@
logger=com.alibaba.otter.canal.client.adapter.logger.LoggerAdapterExample

21
client-adapter/pom.xml Normal file
View File

@@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>canal</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client-adapter</artifactId>
<packaging>pom</packaging>
<name>canal client adapter module for otter ${project.version}</name>
<modules>
<module>common</module>
<module>logger</module>
<module>hbase</module>
</modules>
</project>

View File

@@ -13,6 +13,12 @@
<packaging>jar</packaging> <packaging>jar</packaging>
<name>canal client launcher module for otter ${project.version}</name> <name>canal client launcher module for otter ${project.version}</name>
<dependencies> <dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency> <dependency>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId> <artifactId>canal.client</artifactId>
@@ -24,16 +30,25 @@
<artifactId>canal.kafka.client</artifactId> <artifactId>canal.kafka.client</artifactId>
<version>${project.version}</version> <version>${project.version}</version>
</dependency> </dependency>
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.9.4</version>
</dependency>
<dependency> <dependency>
<groupId>org.yaml</groupId> <groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId> <artifactId>snakeyaml</artifactId>
<version>1.17</version> <version>1.17</version>
</dependency> </dependency>
<!-- outer adapter -->
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.logger</artifactId>
<version>${project.version}</version>
<classifier>jar-with-dependencies</classifier>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.hbase</artifactId>
<version>${project.version}</version>
<classifier>jar-with-dependencies</classifier>
</dependency>
</dependencies> </dependencies>
<build> <build>

View File

@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.client; package com.alibaba.otter.canal.client;
import com.alibaba.otter.canal.client.support.CanalClientConfig; import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
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;
@@ -9,8 +10,9 @@ import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream; import java.io.FileInputStream;
public class ClientLauncher { public class ClientLauncher {
private static final String CLASSPATH_URL_PREFIX = "classpath:"; private static final String CLASSPATH_URL_PREFIX = "classpath:";
private static final Logger logger = LoggerFactory.getLogger(ClientLauncher.class); private static final Logger logger = LoggerFactory.getLogger(ClientLauncher.class);
public static void main(String[] args) { public static void main(String[] args) {
try { try {
@@ -23,7 +25,7 @@ public class ClientLauncher {
if (conf.startsWith(CLASSPATH_URL_PREFIX)) { if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
canalClientConfig = new Yaml().loadAs(ClientLauncher.class.getClassLoader().getResourceAsStream(conf), canalClientConfig = new Yaml().loadAs(ClientLauncher.class.getClassLoader().getResourceAsStream(conf),
CanalClientConfig.class); CanalClientConfig.class);
} else { } else {
canalClientConfig = new Yaml().loadAs(new FileInputStream(conf), CanalClientConfig.class); canalClientConfig = new Yaml().loadAs(new FileInputStream(conf), CanalClientConfig.class);
} }

View File

@@ -1,156 +0,0 @@
package com.alibaba.otter.canal.client.adapter;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.support.Dml;
import com.alibaba.otter.canal.client.support.JdbcTypeUtil;
import com.alibaba.otter.canal.protocol.CanalEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
/**
* 适配器工作线程抽象类
*
* @author machengyuan 2018-8-19 下午11:30:49
* @version 1.0.0
*/
public abstract class AbstractCanalAdapterWorker {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected String canalDestination; // canal实例
protected List<List<CanalOuterAdapter>> canalOuterAdapters; // 外部适配器
protected ExecutorService groupInnerExecutorService; // 组内工作线程池
protected volatile boolean running = false; // 是否运行中
protected Thread thread = null;
protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
logger.error("parse events has an error", e);
}
};
/**
* 将每个entry转换为DML操作对象并调用所有适配器写入
*
* @param entries
*/
protected void convertAndWrite(List<CanalEntry.Entry> entries) {
for (CanalEntry.Entry entry : entries) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
CanalEntry.RowChange rowChange;
try {
rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
}
CanalEntry.EventType eventType = rowChange.getEventType();
final Dml dml = new Dml();
dml.setCanalDestination(canalDestination);
dml.setDatabase(entry.getHeader().getSchemaName());
dml.setTable(entry.getHeader().getTableName());
dml.setType(eventType.toString());
dml.setTs(System.currentTimeMillis());
dml.setSql(rowChange.getSql());
List<Map<String, Object>> data = new ArrayList<>();
List<Map<String, Object>> old = new ArrayList<>();
if (!rowChange.getIsDdl()) {
Set<String> updateSet = new HashSet<>();
for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
Map<String, Object> row = new LinkedHashMap<>();
List<CanalEntry.Column> columns;
if (eventType == CanalEntry.EventType.DELETE) {
columns = rowData.getBeforeColumnsList();
} else {
columns = rowData.getAfterColumnsList();
}
for (CanalEntry.Column column : columns) {
row.put(column.getName(), JdbcTypeUtil.typeConvert(dml.getTable(), column.getName(), column.getValue(),
column.getSqlType(), column.getMysqlType()));
//获取update为true的字段
if (column.getUpdated()) {
updateSet.add(column.getName());
}
}
if (!row.isEmpty()) {
data.add(row);
}
if (eventType == CanalEntry.EventType.UPDATE) {
Map<String, Object> rowOld = new LinkedHashMap<>();
for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
if (updateSet.contains(column.getName())) {
rowOld.put(column.getName(), JdbcTypeUtil.typeConvert(dml.getTable(), column.getName(),
column.getValue(), column.getSqlType(), column.getMysqlType()));
}
}
// update操作将记录修改前的值
if (!rowOld.isEmpty()) {
old.add(rowOld);
}
}
}
if (!data.isEmpty()) {
dml.setData(data);
}
if (!old.isEmpty()) {
dml.setOld(old);
}
}
List<Future<Boolean>> futures = new ArrayList<>();
// 组间适配器并行运行
for (List<CanalOuterAdapter> outerAdapters : canalOuterAdapters) {
final List<CanalOuterAdapter> adapters = outerAdapters;
futures.add(groupInnerExecutorService.submit(
new Callable<Boolean>() {
@Override
public Boolean call() {
boolean flag = true;
// 组内适配器穿行运行,尽量不要配置组内适配器
for (CanalOuterAdapter c : adapters) {
long begin = System.currentTimeMillis();
if (!c.writeOut(dml)) {
logger.error("{} write fail! data: {}", c.getClass().getName(), JSON.toJSONString(dml));
flag = false;
}
if (logger.isDebugEnabled()) {
logger.debug("{} elapsed time: {}", c.getClass().getName(), (System.currentTimeMillis() - begin));
}
}
return flag;
}
})
);
// 等待所有适配器写入完成
// 由于是组间并发操作,所以将阻塞直到耗时最久的工作组操作完成
for (Future<Boolean> f : futures) {
try {
if (!f.get()) {
logger.error("Outer adapter write failed");
}
} catch (InterruptedException | ExecutionException e) {
logger.error(e.getMessage(), e);
}
}
}
}
}
public abstract void start();
public abstract void stop();
}

View File

@@ -0,0 +1,83 @@
package com.alibaba.otter.canal.client.adapter.loader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.protocol.Message;
/**
* 适配器工作线程抽象类
*
* @author machengyuan 2018-8-19 下午11:30:49
* @version 1.0.0
*/
public abstract class AbstractCanalAdapterWorker {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected String canalDestination; // canal实例
protected List<List<CanalOuterAdapter>> canalOuterAdapters; // 外部适配器
protected ExecutorService groupInnerExecutorService; // 组内工作线程池
protected volatile boolean running = false; // 是否运行中
protected Thread thread = null;
protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
logger.error("parse events has an error", e);
}
};
protected void writeOut(final Message message) {
List<Future<Boolean>> futures = new ArrayList<>();
// 组间适配器并行运行
for (List<CanalOuterAdapter> outerAdapters : canalOuterAdapters) {
final List<CanalOuterAdapter> adapters = outerAdapters;
futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
@Override
public Boolean call() {
try {
// 组内适配器穿行运行,尽量不要配置组内适配器
for (CanalOuterAdapter c : adapters) {
long begin = System.currentTimeMillis();
c.writeOut(message);
if (logger.isDebugEnabled()) {
logger.debug("{} elapsed time: {}",
c.getClass().getName(),
(System.currentTimeMillis() - begin));
}
}
return true;
} catch (Exception e) {
return false;
}
}
}));
// 等待所有适配器写入完成
// 由于是组间并发操作,所以将阻塞直到耗时最久的工作组操作完成
for (Future<Boolean> f : futures) {
try {
if (!f.get()) {
logger.error("Outer adapter write failed");
}
} catch (InterruptedException | ExecutionException e) {
logger.error(e.getMessage(), e);
}
}
}
}
public abstract void start();
public abstract void stop();
}

View File

@@ -1,5 +1,7 @@
package com.alibaba.otter.canal.client.adapter; package com.alibaba.otter.canal.client.adapter.loader;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector; import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors; import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
@@ -13,11 +15,13 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
private KafkaCanalConnector connector; private KafkaCanalConnector connector;
private String topic; private String topic;
public CanalAdapterKafkaWorker(String zkServers, String bootstrapServers, String topic, String groupId, List<List<CanalOuterAdapter>> canalOuterAdapters) { public CanalAdapterKafkaWorker(String zkServers, String bootstrapServers, String topic, String groupId,
List<List<CanalOuterAdapter>> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
this.topic = topic; this.topic = topic;
@@ -25,13 +29,14 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector = KafkaCanalConnectors.newKafkaConnector(zkServers, bootstrapServers, topic, null, groupId); connector = KafkaCanalConnectors.newKafkaConnector(zkServers, bootstrapServers, topic, null, groupId);
// connector.setSessionTimeout(5L, TimeUnit.MINUTES); // connector.setSessionTimeout(5L, TimeUnit.MINUTES);
// super.initSwitcher(topic); // super.initSwitcher(topic);
} }
@Override @Override
public void start() { public void start() {
if (!running) { if (!running) {
thread = new Thread(new Runnable() { thread = new Thread(new Runnable() {
@Override @Override
public void run() { public void run() {
process(); process();
@@ -53,9 +58,9 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector.stopRunning(); connector.stopRunning();
running = false; running = false;
// if (switcher != null && !switcher.state()) { // if (switcher != null && !switcher.state()) {
// switcher.set(true); // switcher.set(true);
// } // }
if (thread != null) { if (thread != null) {
try { try {
@@ -78,7 +83,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
} }
private void process() { private void process() {
while (!running) ; while (!running)
;
ExecutorService executor = Executors.newFixedThreadPool(1); ExecutorService executor = Executors.newFixedThreadPool(1);
final AtomicBoolean executing = new AtomicBoolean(true); final AtomicBoolean executing = new AtomicBoolean(true);
while (running) { while (running) {
@@ -90,37 +96,45 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
while (running) { while (running) {
try { try {
// switcher.get(); //等待开关开启 // switcher.get(); //等待开关开启
final Message message = connector.getWithoutAck(); final Message message = connector.getWithoutAck();
executing.set(true); executing.set(true);
if (message != null) { if (message != null) {
executor.submit( executor.submit(new Runnable() {
new Runnable() {
@Override @Override
public void run() { public void run() {
try { try {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size()); logger.debug("topic: {} batchId: {} batchSize: {} ",
} topic,
long begin = System.currentTimeMillis(); message.getId(),
convertAndWrite(message.getEntries()); message.getEntries().size());
long now = System.currentTimeMillis();
if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
}
if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
executing.compareAndSet(true, false);
}
} }
long begin = System.currentTimeMillis();
writeOut(message);
long now = System.currentTimeMillis();
if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
logger.error("topic: {} batchId {} elapsed time: {} ms",
topic,
message.getId(),
now - begin);
}
if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId {} elapsed time: {} ms",
topic,
message.getId(),
now - begin);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
executing.compareAndSet(true, false);
} }
); }
});
while (executing.get()) { // keeping kafka client active while (executing.get()) { // keeping kafka client active
connector.ack(); connector.ack();

View File

@@ -1,11 +1,9 @@
package com.alibaba.otter.canal.client; package com.alibaba.otter.canal.client.adapter.loader;
import com.alibaba.otter.canal.client.adapter.CanalAdapterKafkaWorker;
import com.alibaba.otter.canal.client.adapter.CanalAdapterWorker;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import com.alibaba.otter.canal.client.support.CanalClientConfig; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.support.ExtensionLoader; import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -25,17 +23,18 @@ import java.util.concurrent.Executors;
* @version 1.0.0 * @version 1.0.0
*/ */
public class CanalAdapterLoader { public class CanalAdapterLoader {
private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class);
private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class);
private CanalClientConfig canalClientConfig; private CanalClientConfig canalClientConfig;
private Map<String, CanalAdapterWorker> canalWorkers = new HashMap<>(); private Map<String, CanalAdapterWorker> canalWorkers = new HashMap<>();
private Map<String, CanalAdapterKafkaWorker> canalKafkaWorkers = new HashMap<>(); private Map<String, CanalAdapterKafkaWorker> canalKafkaWorkers = new HashMap<>();
private ExtensionLoader<CanalOuterAdapter> loader; private ExtensionLoader<CanalOuterAdapter> loader;
public CanalAdapterLoader(CanalClientConfig canalClientConfig) { public CanalAdapterLoader(CanalClientConfig canalClientConfig){
this.canalClientConfig = canalClientConfig; this.canalClientConfig = canalClientConfig;
} }
@@ -48,7 +47,8 @@ public class CanalAdapterLoader {
throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics"); throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics");
} }
loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, "" /*TODO canalClientConfig.getClassloaderPolicy()*/); loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class,
"" /* TODO canalClientConfig.getClassloaderPolicy() */);
String canalServerHost = this.canalClientConfig.getCanalServerHost(); String canalServerHost = this.canalClientConfig.getCanalServerHost();
SocketAddress sa = null; SocketAddress sa = null;
@@ -91,20 +91,23 @@ public class CanalAdapterLoader {
List<CanalOuterAdapter> canalOuterAdapters = new ArrayList<>(); List<CanalOuterAdapter> canalOuterAdapters = new ArrayList<>();
for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) { for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) {
// for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) { // for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) {
loadConnector(config, canalOuterAdapters); loadConnector(config, canalOuterAdapters);
// } // }
} }
canalOuterAdapterGroups.add(canalOuterAdapters); canalOuterAdapterGroups.add(canalOuterAdapters);
String zkServers = canalClientConfig.getZookeeperHosts(); String zkServers = canalClientConfig.getZookeeperHosts();
CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkServers, CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkServers,
canalClientConfig.getBootstrapServers(), canalClientConfig.getBootstrapServers(),
kafkaTopic.getTopic(), group.getGroupId(), canalOuterAdapterGroups); kafkaTopic.getTopic(),
group.getGroupId(),
canalOuterAdapterGroups);
canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker); canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker);
canalKafkaWorker.start(); canalKafkaWorker.start();
logger.info("Start adapter for canal-client kafka topic: {} succeed", kafkaTopic.getTopic() + "-" + group.getGroupId()); logger.info("Start adapter for canal-client kafka topic: {} succeed",
kafkaTopic.getTopic() + "-" + group.getGroupId());
} }
} }
} }
@@ -125,8 +128,7 @@ public class CanalAdapterLoader {
} }
/** /**
* 销毁所有适配器 * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁
* 为防止canal实例太多造成销毁阻塞, 并行销毁
*/ */
public void destroy() { public void destroy() {
if (canalWorkers.size() > 0) { if (canalWorkers.size() > 0) {
@@ -134,6 +136,7 @@ public class CanalAdapterLoader {
for (CanalAdapterWorker v : canalWorkers.values()) { for (CanalAdapterWorker v : canalWorkers.values()) {
final CanalAdapterWorker caw = v; final CanalAdapterWorker caw = v;
stopExecutorService.submit(new Runnable() { stopExecutorService.submit(new Runnable() {
@Override @Override
public void run() { public void run() {
caw.stop(); caw.stop();
@@ -147,6 +150,7 @@ public class CanalAdapterLoader {
for (CanalAdapterKafkaWorker v : canalKafkaWorkers.values()) { for (CanalAdapterKafkaWorker v : canalKafkaWorkers.values()) {
final CanalAdapterKafkaWorker cakw = v; final CanalAdapterKafkaWorker cakw = v;
stopKafkaExecutorService.submit(new Runnable() { stopKafkaExecutorService.submit(new Runnable() {
@Override @Override
public void run() { public void run() {
cakw.stop(); cakw.stop();

View File

@@ -1,7 +1,9 @@
package com.alibaba.otter.canal.client.adapter; package com.alibaba.otter.canal.client.adapter.loader;
import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker;
import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
@@ -16,19 +18,21 @@ import java.util.concurrent.Executors;
* @version 1.0.0 * @version 1.0.0
*/ */
public class CanalAdapterWorker extends AbstractCanalAdapterWorker { public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
private static final int BATCH_SIZE = 50; private static final int BATCH_SIZE = 50;
private static final int SO_TIMEOUT = 0; private static final int SO_TIMEOUT = 0;
private CanalConnector connector; private CanalConnector connector;
/** /**
* 单台client适配器worker的构造方法 * 单台client适配器worker的构造方法
* *
* @param canalDestination canal实例名 * @param canalDestination canal实例名
* @param address canal-server地址 * @param address canal-server地址
* @param canalOuterAdapters 外部适配器组 * @param canalOuterAdapters 外部适配器组
*/ */
public CanalAdapterWorker(String canalDestination, SocketAddress address, List<List<CanalOuterAdapter>> canalOuterAdapters) { public CanalAdapterWorker(String canalDestination, SocketAddress address,
List<List<CanalOuterAdapter>> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.canalDestination = canalDestination; this.canalDestination = canalDestination;
groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
@@ -38,24 +42,26 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
/** /**
* HA模式下client适配器worker的构造方法 * HA模式下client适配器worker的构造方法
* *
* @param canalDestination canal实例名 * @param canalDestination canal实例名
* @param zookeeperHosts zookeeper地址 * @param zookeeperHosts zookeeper地址
* @param canalOuterAdapters 外部适配器组 * @param canalOuterAdapters 外部适配器组
*/ */
public CanalAdapterWorker(String canalDestination, String zookeeperHosts, List<List<CanalOuterAdapter>> canalOuterAdapters) { public CanalAdapterWorker(String canalDestination, String zookeeperHosts,
List<List<CanalOuterAdapter>> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.canalDestination = canalDestination; this.canalDestination = canalDestination;
groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", ""); connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", "");
((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT); ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT);
// super.initSwitcher(canalDestination); // super.initSwitcher(canalDestination);
} }
@Override @Override
public void start() { public void start() {
if (!running) { if (!running) {
thread = new Thread(new Runnable() { thread = new Thread(new Runnable() {
@Override @Override
public void run() { public void run() {
process(); process();
@@ -74,9 +80,9 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
return; return;
} }
// if (switcher != null && !switcher.state()) { // if (switcher != null && !switcher.state()) {
// switcher.set(true); // switcher.set(true);
// } // }
connector.stopRunning(); connector.stopRunning();
running = false; running = false;
@@ -103,27 +109,29 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
} }
private void process() { private void process() {
while (!running) ; //waiting until running == true while (!running)
; // waiting until running == true
while (running) { while (running) {
try { try {
// if (switcher != null) { // if (switcher != null) {
// switcher.get(); // switcher.get();
// } // }
logger.info("=============> Start to connect destination: {} <=============", this.canalDestination); logger.info("=============> Start to connect destination: {} <=============", this.canalDestination);
connector.connect(); connector.connect();
logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination); logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination);
connector.subscribe(); connector.subscribe();
logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination); logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination);
while (running) { while (running) {
// try { // try {
// if (switcher != null) { // if (switcher != null) {
// switcher.get(); // switcher.get();
// } // }
// } catch (TimeoutException e) { // } catch (TimeoutException e) {
// break; // break;
// } // }
// server配置canal.instance.network.soTimeout(默认: 30s) 范围内未与server交互,server将关闭本次socket连接 // server配置canal.instance.network.soTimeout(默认: 30s)
// 范围内未与server交互,server将关闭本次socket连接
Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据 Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据
long batchId = message.getId(); long batchId = message.getId();
try { try {
@@ -133,17 +141,23 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
try { try {
Thread.sleep(1000); Thread.sleep(1000);
} catch (InterruptedException e) { } catch (InterruptedException e) {
//ignore // ignore
} }
} else { } else {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} batchSize: {} ", this.canalDestination, batchId, size); logger.debug("destination: {} batchId: {} batchSize: {} ",
this.canalDestination,
batchId,
size);
} }
long begin = System.currentTimeMillis(); long begin = System.currentTimeMillis();
convertAndWrite(message.getEntries()); writeOut(message);
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} elapsed time: {} ms", this.canalDestination, batchId, now - begin); logger.debug("destination: {} batchId: {} elapsed time: {} ms",
this.canalDestination,
batchId,
now - begin);
} }
} }
connector.ack(batchId); // 提交确认 connector.ack(batchId); // 提交确认
@@ -153,7 +167,6 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error("process error!", e); logger.error("process error!", e);
} finally { } finally {
@@ -161,11 +174,11 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
logger.info("=============> Disconnect destination: {} <=============", this.canalDestination); logger.info("=============> Disconnect destination: {} <=============", this.canalDestination);
} }
if (running) { //is reconnect if (running) { // is reconnect
try { try {
Thread.sleep(1000); Thread.sleep(1000);
} catch (InterruptedException e) { } catch (InterruptedException e) {
//ignore // ignore
} }
} }

View File

@@ -1,36 +0,0 @@
package com.alibaba.otter.canal.client.example;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.support.Dml;
import com.alibaba.otter.canal.client.support.SPI;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 外部适配器示例
*
* @author machengyuan 2018-8-19 下午11:45:38
* @version 1.0.0
*/
@SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name
public class LoggerAdapterExample implements CanalOuterAdapter {
private Logger logger = LoggerFactory.getLogger(this.getClass());
@Override
public Boolean writeOut(Dml dml) {
// 直接输出一个日志信息
logger.info(dml.toString());
return true;
}
@Override
public void init(CanalOuterAdapterConfiguration configuration) {
}
@Override
public void destroy() {
}
}

View File

@@ -1 +0,0 @@
logger=com.alibaba.otter.canal.client.example.LoggerAdapterExample

View File

@@ -1,5 +1,5 @@
#canalServerHost: 127.0.0.1:11111 canalServerHost: 127.0.0.1:11111
zookeeperHosts: 127.0.0.1:2181 #zookeeperHosts: 127.0.0.1:2181
#bootstrapServers: kafka1.mytest.com:9092,kafka2.mytest.com:9092 #bootstrapServers: kafka1.mytest.com:9092,kafka2.mytest.com:9092
canalInstances: canalInstances:
@@ -7,9 +7,9 @@ canalInstances:
adapterGroups: adapterGroups:
- outAdapters: - outAdapters:
- name: logger - name: logger
# - name: hbase - name: hbase
# hosts: hosts: slave1:2181
# properties: {znodeParent: /hbase} properties: {znodeParent: "/hbase-unsecure"}
#kafkaTopics: #kafkaTopics:
#- topic: devmysql4308 #- topic: devmysql4308

View File

@@ -36,12 +36,12 @@
<logger name="com.alibaba.otter.canal.client" additivity="false"> <logger name="com.alibaba.otter.canal.client" additivity="false">
<level value="INFO" /> <level value="INFO" />
<appender-ref ref="STDOUT"/> <!--<appender-ref ref="STDOUT"/>-->
<appender-ref ref="CANAL-ROOT" /> <appender-ref ref="CANAL-ROOT" />
</logger> </logger>
<root level="WARN"> <root level="WARN">
<appender-ref ref="STDOUT"/> <!--<appender-ref ref="STDOUT"/>-->
<appender-ref ref="CANAL-ROOT" /> <appender-ref ref="CANAL-ROOT" />
</root> </root>
</configuration> </configuration>

View File

@@ -120,6 +120,7 @@
<module>kafka</module> <module>kafka</module>
<module>kafka-client</module> <module>kafka-client</module>
<module>prometheus</module> <module>prometheus</module>
<module>client-adapter</module>
<module>client-launcher</module> <module>client-launcher</module>
</modules> </modules>
@@ -333,7 +334,7 @@
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId> <artifactId>maven-compiler-plugin</artifactId>
<version>3.8.0</version> <version>3.7.0</version>
<configuration> <configuration>
<source>${java_source_version}</source> <source>${java_source_version}</source>
<target>${java_target_version}</target> <target>${java_target_version}</target>