diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml new file mode 100644 index 00000000..a3a78073 --- /dev/null +++ b/client-adapter/common/pom.xml @@ -0,0 +1,33 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.0-SNAPSHOT + + 4.0.0 + com.alibaba.otter + client-adapter.common + jar + canal client adapter common module for otter ${project.version} + + + com.alibaba.otter + canal.protocol + ${project.version} + + + org.slf4j + slf4j-api + 1.7.12 + + + joda-time + joda-time + 2.9.4 + + + + \ No newline at end of file diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java new file mode 100644 index 00000000..53368472 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java @@ -0,0 +1,34 @@ +package com.alibaba.otter.canal.client.adapter; + +import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.SPI; +import com.alibaba.otter.canal.protocol.Message; + +/** + * 外部适配器接口 + * + * @author machengyuan 2018-8-18 下午10:14:02 + * @version 1.0.0 + */ +@SPI("logger") +public interface CanalOuterAdapter { + + /** + * 外部适配器初始化接口 + * + * @param configuration 外部适配器配置信息 + */ + void init(CanalOuterAdapterConfiguration configuration); + + /** + * 往适配器中写入数据 + * + * @param message message数据包 + */ + void writeOut(Message message); + + /** + * 外部适配器销毁接口 + */ + void destroy(); +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java new file mode 100644 index 00000000..4f8ff396 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -0,0 +1,181 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +/** + * 配置信息类 + * + * @author machengyuan 2018-8-18 下午10:40:12 + * @version 1.0.0 + */ +public class CanalClientConfig { + + private String canalServerHost; + + private String zookeeperHosts; + + private Properties properties; + + private String bootstrapServers; + + private List kafkaTopics = new ArrayList<>(); + + private List canalInstances = new ArrayList<>(); + + public String getCanalServerHost() { + return canalServerHost; + } + + public void setCanalServerHost(String canalServerHost) { + this.canalServerHost = canalServerHost; + } + + public String getZookeeperHosts() { + return zookeeperHosts; + } + + public void setZookeeperHosts(String zookeeperHosts) { + this.zookeeperHosts = zookeeperHosts; + } + + public Properties getProperties() { + return properties; + } + + public void setProperties(Properties properties) { + this.properties = properties; + } + + public String getBootstrapServers() { + return bootstrapServers; + } + + public void setBootstrapServers(String bootstrapServers) { + this.bootstrapServers = bootstrapServers; + } + + public List getKafkaTopics() { + return kafkaTopics; + } + + public void setKafkaTopics(List kafkaTopics) { + this.kafkaTopics = kafkaTopics; + } + + public List getCanalInstances() { + return canalInstances; + } + + public void setCanalInstances(List canalInstances) { + this.canalInstances = canalInstances; + } + + public static class CanalInstance { + + private String instance; + + private List adapterGroups; + + public String getInstance() { + return instance; + } + + public void setInstance(String instance) { + if (instance != null) { + this.instance = instance.trim(); + } + } + + public List getAdapterGroups() { + return adapterGroups; + } + + public void setAdapterGroups(List adapterGroups) { + this.adapterGroups = adapterGroups; + } + } + + public static class AdapterGroup { + + private List outAdapters; + + public List getOutAdapters() { + return outAdapters; + } + + public void setOutAdapters(List outAdapters) { + this.outAdapters = outAdapters; + } + } + + public static class KafkaTopic { + + private String topic; + + private List groups = new ArrayList<>(); + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public List getGroups() { + return groups; + } + + public void setGroups(List groups) { + this.groups = groups; + } + } + + public static class Group { + + private String groupId; + + // private List adapters = new ArrayList<>(); + + private List outAdapters; + + public String getGroupId() { + return groupId; + } + + public void setGroupId(String groupId) { + this.groupId = groupId; + } + + public List getOutAdapters() { + return outAdapters; + } + + public void setOutAdapters(List outAdapters) { + this.outAdapters = outAdapters; + } + + // public List getAdapters() { + // return adapters; + // } + // + // public void setAdapters(List adapters) { + // this.adapters = adapters; + // } + } + + // public static class Adaptor { + // private List outAdapters; + // + // public List getOutAdapters() { + // return outAdapters; + // } + // + // public void setOutAdapters(List outAdapters) + // { + // this.outAdapters = outAdapters; + // } + // } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java new file mode 100644 index 00000000..e6e34b67 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java @@ -0,0 +1,52 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.util.Properties; + +/** + * 外部适配器配置信息类 + * + * @author machengyuan 2018-8-18 下午10:15:12 + * @version 1.0.0 + */ +public class CanalOuterAdapterConfiguration { + + private String name; // 适配器名称, 如: logger, hbase, es + + private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址 + + private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址 + + private Properties properties; // 其余参数, 可填写适配器中的所需的配置信息 + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getHosts() { + return hosts; + } + + public void setHosts(String hosts) { + this.hosts = hosts; + } + + public Properties getProperties() { + return properties; + } + + public void setProperties(Properties properties) { + this.properties = properties; + } + + public String getZkHosts() { + return zkHosts; + } + + public void setZkHosts(String zkHosts) { + this.zkHosts = zkHosts; + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java new file mode 100644 index 00000000..fe7333e5 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java @@ -0,0 +1,96 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.io.Serializable; +import java.util.List; +import java.util.Map; + +/** + * DML操作转换对象 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class Dml implements Serializable { + + private static final long serialVersionUID = 2611556444074013268L; + + private String database; + private String table; + private String type; + private Long ts; + private String sql; + private List> data; + private List> old; + + 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 getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public Long getTs() { + return ts; + } + + public void setTs(Long ts) { + this.ts = ts; + } + + public String getSql() { + return sql; + } + + public void setSql(String sql) { + this.sql = sql; + } + + public List> getData() { + return data; + } + + public void setData(List> data) { + this.data = data; + } + + public List> getOld() { + return old; + } + + public void setOld(List> old) { + this.old = old; + } + + public void clear() { + database = null; + table = null; + type = null; + ts = null; + data = null; + old = null; + sql = null; + } + + @Override + public String toString() { + return "Dml{" + "database='" + database + '\'' + ", table='" + table + '\'' + ", type='" + type + '\'' + ", ts=" + + ts + ", sql='" + sql + '\'' + ", data=" + data + ", old=" + old + '}'; + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java new file mode 100644 index 00000000..01b05054 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java @@ -0,0 +1,563 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.io.*; +import java.net.MalformedURLException; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.file.Paths; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.regex.Pattern; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * SPI 类加载器 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class ExtensionLoader { + + private static final Logger logger = LoggerFactory + .getLogger(ExtensionLoader.class); + + private static final String SERVICES_DIRECTORY = "META-INF/services/"; + + private static final String CANAL_DIRECTORY = "META-INF/canal/"; + + private static final String DEFAULT_CLASSLOADER_POLICY = "internal"; + + private static final Pattern NAME_SEPARATOR = Pattern + .compile("\\s*[,]+\\s*"); + + private static final ConcurrentMap, ExtensionLoader> EXTENSION_LOADERS = new ConcurrentHashMap<>(); + + private static final ConcurrentMap, Object> EXTENSION_INSTANCES = new ConcurrentHashMap<>(); + + private final Class type; + + private final String classLoaderPolicy; + + private final ConcurrentMap, String> cachedNames = new ConcurrentHashMap<>(); + + private final Holder>> cachedClasses = new Holder<>(); + + private final ConcurrentMap> cachedInstances = new ConcurrentHashMap<>(); + + private String cachedDefaultName; + + private ConcurrentHashMap exceptions = new ConcurrentHashMap<>(); + + private static boolean withExtensionAnnotation(Class type) { + return type.isAnnotationPresent(SPI.class); + } + + public static ExtensionLoader getExtensionLoader(Class type) { + return getExtensionLoader(type, DEFAULT_CLASSLOADER_POLICY); + } + + @SuppressWarnings("unchecked") + public static ExtensionLoader getExtensionLoader(Class type, String classLoaderPolicy) { + if (type == null) throw new IllegalArgumentException("Extension type == null"); + if (!type.isInterface()) { + throw new IllegalArgumentException("Extension type(" + type + ") is not interface!"); + } + if (!withExtensionAnnotation(type)) { + throw new IllegalArgumentException("Extension type(" + type + ") is not extension, because WITHOUT @" + + SPI.class.getSimpleName() + " Annotation!"); + } + + ExtensionLoader loader = (ExtensionLoader) EXTENSION_LOADERS.get(type); + if (loader == null) { + EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader(type, classLoaderPolicy)); + loader = (ExtensionLoader) EXTENSION_LOADERS.get(type); + } + return loader; + } + + private ExtensionLoader(Class type, String classLoaderPolicy){ + this.type = type; + this.classLoaderPolicy = classLoaderPolicy; + } + + public String getExtensionName(T extensionInstance) { + return getExtensionName(extensionInstance.getClass()); + } + + public String getExtensionName(Class extensionClass) { + return cachedNames.get(extensionClass); + } + + public ConcurrentHashMap getExceptions() { + return exceptions; + } + + /** + * 返回扩展点实例，如果没有指定的扩展点或是还没加载（即实例化）则返回null注意：此方法不会触发扩展点的加载 + *

+ * 一般应该调用{@link #getExtension(String)}方法获得扩展，这个方法会触发扩展点加载 + * + * @see #getExtension(String) + */ + @SuppressWarnings("unchecked") + public T getLoadedExtension(String name) { + if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null"); + Holder holder = cachedInstances.get(name); + if (holder == null) { + cachedInstances.putIfAbsent(name, new Holder<>()); + holder = cachedInstances.get(name); + } + return (T) holder.get(); + } + + /** + * 返回已经加载的扩展点的名字 + *

+ * 一般应该调用{@link #getSupportedExtensions()}方法获得扩展，这个方法会返回所有的扩展点 + * + * @see #getSupportedExtensions() + */ + public Set getLoadedExtensions() { + return Collections.unmodifiableSet(new TreeSet<>(cachedInstances.keySet())); + } + + /** + * 返回指定名字的扩展 + * + * @param name + * @return + */ + @SuppressWarnings("unchecked") + public T getExtension(String name) { + if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null"); + if ("true".equals(name)) { + return getDefaultExtension(); + } + Holder holder = cachedInstances.get(name); + if (holder == null) { + cachedInstances.putIfAbsent(name, new Holder<>()); + holder = cachedInstances.get(name); + } + Object instance = holder.get(); + if (instance == null) { + synchronized (holder) { + instance = holder.get(); + if (instance == null) { + instance = createExtension(name); + holder.set(instance); + } + } + } + return (T) instance; + } + + /** + * 返回缺省的扩展，如果没有设置则返回null + */ + public T getDefaultExtension() { + getExtensionClasses(); + if (null == cachedDefaultName || cachedDefaultName.length() == 0 || "true".equals(cachedDefaultName)) { + return null; + } + return getExtension(cachedDefaultName); + } + + public boolean hasExtension(String name) { + if (name == null || name.length() == 0) throw new IllegalArgumentException("Extension name == null"); + try { + return getExtensionClass(name) != null; + } catch (Throwable t) { + return false; + } + } + + public Set getSupportedExtensions() { + Map> clazzes = getExtensionClasses(); + return Collections.unmodifiableSet(new TreeSet(clazzes.keySet())); + } + + /** + * 返回缺省的扩展点名，如果没有设置缺省则返回null + */ + public String getDefaultExtensionName() { + getExtensionClasses(); + return cachedDefaultName; + } + + /** + * 编程方式添加新扩展点 + * + * @param name 扩展点名 + * @param clazz 扩展点类 + * @throws IllegalStateException 要添加扩展点名已经存在 + */ + public void addExtension(String name, Class clazz) { + getExtensionClasses(); // load classes + + if (!type.isAssignableFrom(clazz)) { + throw new IllegalStateException("Input type " + clazz + "not implement Extension " + type); + } + if (clazz.isInterface()) { + throw new IllegalStateException("Input type " + clazz + "can not be interface!"); + } + + if (name == null || "".equals(name)) { + throw new IllegalStateException("Extension name is blank (Extension " + type + ")!"); + } + if (cachedClasses.get().containsKey(name)) { + throw new IllegalStateException("Extension name " + name + " already existed(Extension " + type + ")!"); + } + + cachedNames.put(clazz, name); + cachedClasses.get().put(name, clazz); + } + + /** + * 编程方式添加替换已有扩展点 + * + * @param name 扩展点名 + * @param clazz 扩展点类 + * @throws IllegalStateException 要添加扩展点名已经存在 + * @deprecated 不推荐应用使用，一般只在测试时可以使用 + */ + @Deprecated + public void replaceExtension(String name, Class clazz) { + getExtensionClasses(); // load classes + + if (!type.isAssignableFrom(clazz)) { + throw new IllegalStateException("Input type " + clazz + "not implement Extension " + type); + } + if (clazz.isInterface()) { + throw new IllegalStateException("Input type " + clazz + "can not be interface!"); + } + + if (name == null || "".equals(name)) { + throw new IllegalStateException("Extension name is blank (Extension " + type + ")!"); + } + if (!cachedClasses.get().containsKey(name)) { + throw new IllegalStateException("Extension name " + name + " not existed(Extension " + type + ")!"); + } + + cachedNames.put(clazz, name); + cachedClasses.get().put(name, clazz); + cachedInstances.remove(name); + } + + @SuppressWarnings("unchecked") + private T createExtension(String name) { + Class clazz = getExtensionClasses().get(name); + if (clazz == null) { + throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type + + ") could not be instantiated: class could not be found"); + } + try { + T instance = (T) EXTENSION_INSTANCES.get(clazz); + if (instance == null) { + EXTENSION_INSTANCES.putIfAbsent(clazz, (T) clazz.newInstance()); + instance = (T) EXTENSION_INSTANCES.get(clazz); + } + return instance; + } catch (Throwable t) { + throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type + + ") could not be instantiated: " + t.getMessage(), + t); + } + } + + private Class getExtensionClass(String name) { + if (type == null) throw new IllegalArgumentException("Extension type == null"); + if (name == null) throw new IllegalArgumentException("Extension name == null"); + Class clazz = getExtensionClasses().get(name); + if (clazz == null) + throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName() + "!"); + return clazz; + } + + private Map> getExtensionClasses() { + Map> classes = cachedClasses.get(); + if (classes == null) { + synchronized (cachedClasses) { + classes = cachedClasses.get(); + if (classes == null) { + classes = loadExtensionClasses(); + cachedClasses.set(classes); + } + } + } + return classes; + } + + private String getJarDirectoryPath() { + URL url = Thread.currentThread().getContextClassLoader().getResource(""); + if (url == null) { + throw new IllegalStateException("failed to get class loader resource"); + } + String dirtyPath = url.toString(); + 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("%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. + jarPath = jarPath.replaceAll("/classes/.*", "/classes/"); + } + return Paths.get(jarPath).getParent().toString(); // Paths - from java 8 + } + + private Map> loadExtensionClasses() { + final SPI defaultAnnotation = type.getAnnotation(SPI.class); + if (defaultAnnotation != null) { + String value = defaultAnnotation.value(); + if ((value = value.trim()).length() > 0) { + String[] names = NAME_SEPARATOR.split(value); + if (names.length > 1) { + throw new IllegalStateException("more than 1 default extension name on extension " + type.getName() + + ": " + Arrays.toString(names)); + } + if (names.length == 1) cachedDefaultName = names[0]; + } + } + + Map> extensionClasses = new HashMap>(); + + // 1. lib folder，customized extension classLoader （jar_dir/lib） + String dir = File.separator + this.getJarDirectoryPath() + File.separator + "lib"; + logger.info("extension classpath dir: " + 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()) { + File[] files = externalLibDir.listFiles(new FilenameFilter() { + + @Override + public boolean accept(File dir, String name) { + return name.endsWith(".jar"); + } + }); + if (files != null) { + for (File f : files) { + URL url = null; + try { + url = f.toURI().toURL(); + } catch (MalformedURLException e) { + throw new RuntimeException("load extension jar failed!", e); + } + + ClassLoader parent = Thread.currentThread().getContextClassLoader(); + URLClassLoader localClassLoader; + if (classLoaderPolicy == null || "".equals(classLoaderPolicy) + || DEFAULT_CLASSLOADER_POLICY.equalsIgnoreCase(classLoaderPolicy)) { + localClassLoader = new URLClassLoader(new URL[] { url }, parent) { + + @Override + public Class loadClass(String name) throws ClassNotFoundException { + Class c = findLoadedClass(name); + if (c != null) { + return c; + } + + if (name.startsWith("java.") || name.startsWith("org.slf4j.") + || name.startsWith("org.apache.logging") + || name.startsWith("org.apache.commons.logging.")) { + // || name.startsWith("org.apache.hadoop.")) { + c = super.loadClass(name); + } + if (c != null) return c; + + try { + // 先加载jar内的class，可避免jar冲突 + c = findClass(name); + } catch (ClassNotFoundException e) { + c = null; + } + if (c != null) { + return c; + } + + return super.loadClass(name); + } + + @Override + public Enumeration getResources(String name) throws IOException { + @SuppressWarnings("unchecked") + Enumeration[] tmp = (Enumeration[]) new Enumeration[2]; + + tmp[0] = findResources(name); // local class path first + // tmp[1] = super.getResources(name); + + return new CompoundEnumeration<>(tmp); + } + }; + } else { + localClassLoader = new URLClassLoader(new URL[] { url }, parent); + } + + loadFile(extensionClasses, CANAL_DIRECTORY, localClassLoader); + loadFile(extensionClasses, SERVICES_DIRECTORY, localClassLoader); + } + } + } + // 2. load inner extension class with default classLoader + ClassLoader classLoader = findClassLoader(); + loadFile(extensionClasses, CANAL_DIRECTORY, classLoader); + loadFile(extensionClasses, SERVICES_DIRECTORY, classLoader); + + return extensionClasses; + } + + public static class CompoundEnumeration implements Enumeration { + + private Enumeration[] enums; + private int index = 0; + + public CompoundEnumeration(Enumeration[] 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> extensionClasses, String dir, ClassLoader classLoader) { + String fileName = dir + type.getName(); + try { + Enumeration urls; + if (classLoader != null) { + urls = classLoader.getResources(fileName); + } else { + urls = ClassLoader.getSystemResources(fileName); + } + if (urls != null) { + while (urls.hasMoreElements()) { + URL url = urls.nextElement(); + try { + BufferedReader reader = null; + try { + reader = new BufferedReader(new InputStreamReader(url.openStream(), "utf-8")); + String line = null; + while ((line = reader.readLine()) != null) { + final int ci = line.indexOf('#'); + if (ci >= 0) line = line.substring(0, ci); + line = line.trim(); + if (line.length() > 0) { + try { + String name = null; + int i = line.indexOf('='); + if (i > 0) { + name = line.substring(0, i).trim(); + line = line.substring(i + 1).trim(); + } + if (line.length() > 0) { + Class clazz = classLoader.loadClass(line); + // Class clazz = Class.forName(line, true, classLoader); + if (!type.isAssignableFrom(clazz)) { + throw new IllegalStateException( + "Error when load extension class(interface: " + type + + ", class line: " + clazz.getName() + + "), class " + clazz.getName() + + "is not subtype of interface."); + } else { + try { + clazz.getConstructor(type); + } catch (NoSuchMethodException e) { + clazz.getConstructor(); + String[] names = NAME_SEPARATOR.split(name); + if (names != null && names.length > 0) { + for (String n : names) { + if (!cachedNames.containsKey(clazz)) { + cachedNames.put(clazz, n); + } + Class c = extensionClasses.get(n); + if (c == null) { + extensionClasses.put(n, clazz); + } else if (c != clazz) { + cachedNames.remove(clazz); + throw new IllegalStateException( + "Duplicate extension " + type.getName() + " name " + + n + " on " + + c.getName() + " and " + + clazz.getName()); + } + } + } + } + } + } + } catch (Throwable 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); + } + } + } // end of while read lines + } finally { + if (reader != null) { + reader.close(); + } + } + } catch (Throwable t) { + logger.error("Exception when load extension class(interface: " + type + ", class file: " + url + + ") in " + url, + t); + } + } // end of while urls + } + } catch (Throwable t) { + logger.error( + "Exception when load extension class(interface: " + type + ", description file: " + fileName + ").", + t); + } + } + + private static ClassLoader findClassLoader() { + return ExtensionLoader.class.getClassLoader(); + } + + @Override + public String toString() { + return this.getClass().getName() + "[" + type.getName() + "]"; + } + + private static class Holder { + + private volatile T value; + + private void set(T value) { + this.value = value; + } + + private T get() { + return value; + } + + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java new file mode 100644 index 00000000..fd65589f --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java @@ -0,0 +1,103 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.sql.Date; +import java.sql.Time; +import java.sql.Timestamp; +import java.sql.Types; + +import org.joda.time.DateTime; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 类型转换工具类 + * + * @author machengyuan 2018-8-19 下午06:14:23 + * @version 1.0.0 + */ +public class JdbcTypeUtil { + + private static Logger logger = LoggerFactory.getLogger(JdbcTypeUtil.class); + + public static Object typeConvert(String tableName, String columnName, String value, int sqlType, String mysqlType) { + if (value == null || value.equals("")) { + return null; + } + + try { + Object res; + switch (sqlType) { + case Types.INTEGER: + res = Integer.parseInt(value); + break; + case Types.SMALLINT: + res = Short.parseShort(value); + break; + case Types.BIT: + case Types.TINYINT: + res = Byte.parseByte(value); + break; + case Types.BIGINT: + if (mysqlType.startsWith("bigint") && mysqlType.endsWith("unsigned")) { + res = new BigInteger(value); + } else { + res = Long.parseLong(value); + } + break; + // case Types.BIT: + case Types.BOOLEAN: + res = !"0".equals(value); + break; + case Types.DOUBLE: + case Types.FLOAT: + res = Double.parseDouble(value); + break; + case Types.REAL: + res = Float.parseFloat(value); + break; + case Types.DECIMAL: + case Types.NUMERIC: + res = new BigDecimal(value); + break; + case Types.BINARY: + case Types.VARBINARY: + case Types.LONGVARBINARY: + case Types.BLOB: + res = value.getBytes("ISO-8859-1"); + break; + case Types.DATE: + if (!value.startsWith("0000-00-00")) { + value = value.trim().replace(" ", "T"); + DateTime dt = new DateTime(value); + res = new Date(dt.toDate().getTime()); + } else { + res = null; + } + break; + case Types.TIME: + value = "T" + value; + DateTime dt = new DateTime(value); + res = new Time(dt.toDate().getTime()); + break; + case Types.TIMESTAMP: + if (!value.startsWith("0000-00-00")) { + value = value.trim().replace(" ", "T"); + dt = new DateTime(value); + res = new Timestamp(dt.toDate().getTime()); + } else { + res = null; + } + break; + case Types.CLOB: + default: + res = value; + } + return res; + } catch (Exception e) { + logger.error("table: {} column: {}, failed convert type {} to {}", tableName, columnName, value, sqlType); + return value; + } + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java new file mode 100644 index 00000000..7f3722f1 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -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 consumer) { + if (message == null) { + return; + } + List 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> data = new ArrayList<>(); + List> old = new ArrayList<>(); + + if (!rowChange.getIsDdl()) { + Set updateSet = new HashSet<>(); + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE + && eventType != CanalEntry.EventType.DELETE) { + continue; + } + + Map row = new LinkedHashMap<>(); + List 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 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 { + + void accept(T t); + } +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/SPI.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/SPI.java new file mode 100644 index 00000000..9c0116c6 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/SPI.java @@ -0,0 +1,12 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.lang.annotation.*; + +@Documented +@Retention(RetentionPolicy.RUNTIME) +@Target({ ElementType.TYPE }) +public @interface SPI { + + // Default SPI name + String value() default ""; +} diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml new file mode 100644 index 00000000..a8ec67a0 --- /dev/null +++ b/client-adapter/hbase/pom.xml @@ -0,0 +1,101 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.0-SNAPSHOT + + 4.0.0 + com.alibaba.otter + client-adapter.hbase + jar + canal client adapter hbase module for otter ${project.version} + + + com.alibaba.otter + client-adapter.common + ${project.version} + provided + + + org.yaml + snakeyaml + 1.17 + + + org.apache.hbase + hbase-client + 1.1.2 + + + org.slf4j + slf4j-log4j12 + + + + + com.google.guava + guava + 12.0.1 + + + + + + + maven-jar-plugin + + + true + + + **/hbase-mapping/** + + + + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + + \ No newline at end of file diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java new file mode 100644 index 00000000..03103b8b --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -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 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 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() { + + @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); + } + } + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java new file mode 100644 index 00000000..0e4cbfe6 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java @@ -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 columns; + private ColumnItem rowKeyColumn; + private String etlCondition; + + private Map columnItems = new LinkedHashMap<>(); + private Set 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 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 columns) { + this.columns = columns; + + if (columns != null) { + for (Map.Entry 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 getColumnItems() { + return columnItems; + } + + public void setColumnItems(Map columnItems) { + this.columnItems = columnItems; + } + + public Set getFamilies() { + return families; + } + + public void setFamilies(Set 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; + } + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java new file mode 100644 index 00000000..f00e6282 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java @@ -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表映射配置加载器 + *

+ * 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。 + * 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式 + *

+ * + * @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 load() { + logger.info("## Start loading mapping config ... "); + String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf"); + + Map result = new LinkedHashMap<>(); + + String[] configLines = mappingConfigContent.split("\n"); + for (String c : configLines) { + 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 + } + } + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java new file mode 100644 index 00000000..13f6a3ac --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java @@ -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> 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 rows = new ArrayList<>(); + for (Map 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 data) { + Map columnItems = hbaseOrm.getColumnItems(); + int i = 0; + for (Map.Entry 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> data = dml.getData(); + List> 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 rows = new ArrayList<>(); + out: for (Map 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 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 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> 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 rowKeys = new HashSet<>(); + for (Map 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 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> data = dml.getData(); + List> 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 rowKeys = new HashSet<>(); + List rows = new ArrayList<>(); + for (Map r : data) { + // 拼接老的rowKey + List 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 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(); + } + +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HRow.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HRow.java new file mode 100644 index 00000000..adeb2ff5 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HRow.java @@ -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 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 getCells() { + return cells; + } + + public void setCells(List 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; + } + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java new file mode 100644 index 00000000..62b7ca03 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java @@ -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 rows) { + boolean flag = false; + try { + HTable table = (HTable) conn.getTable(TableName.valueOf(tableName)); + List 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 rowKeys) { + boolean flag = false; + try { + HTable table = (HTable) conn.getTable(TableName.valueOf(tableName)); + List 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; + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java new file mode 100644 index 00000000..09c63a6c --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java @@ -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; + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java new file mode 100644 index 00000000..25473681 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java @@ -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; + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java new file mode 100644 index 00000000..9a567191 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java @@ -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; + } +} diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java new file mode 100644 index 00000000..07aad00f --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java @@ -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 toObject(byte[] bytes, Class 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 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; + } +} diff --git a/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter new file mode 100644 index 00000000..321d49cf --- /dev/null +++ b/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter @@ -0,0 +1 @@ +hbase=com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter \ No newline at end of file diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf new file mode 100644 index 00000000..ff5b292e --- /dev/null +++ b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf @@ -0,0 +1,5 @@ +# 详细映射配置 +mytest_person2.yml + +# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认 +mytest.person \ No newline at end of file diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml new file mode 100644 index 00000000..575a1bf5 --- /dev/null +++ b/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml @@ -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 动态长度 \ No newline at end of file diff --git a/client-adapter/logger/pom.xml b/client-adapter/logger/pom.xml new file mode 100644 index 00000000..e6c5d218 --- /dev/null +++ b/client-adapter/logger/pom.xml @@ -0,0 +1,47 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.0-SNAPSHOT + + 4.0.0 + com.alibaba.otter + client-adapter.logger + jar + canal client adapter logger example module for otter ${project.version} + + + com.alibaba.otter + client-adapter.common + ${project.version} + provided + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + \ No newline at end of file diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java new file mode 100644 index 00000000..c6c1cb44 --- /dev/null +++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java @@ -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() { + + @Override + public void accept(Dml dml) { + logger.info(dml.toString()); + } + }); + } + + @Override + public void init(CanalOuterAdapterConfiguration configuration) { + + } + + @Override + public void destroy() { + + } +} diff --git a/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter new file mode 100644 index 00000000..bfeae48e --- /dev/null +++ b/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter @@ -0,0 +1 @@ +logger=com.alibaba.otter.canal.client.adapter.logger.LoggerAdapterExample \ No newline at end of file diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml new file mode 100644 index 00000000..5398a9c5 --- /dev/null +++ b/client-adapter/pom.xml @@ -0,0 +1,21 @@ + + + + canal + com.alibaba.otter + 1.1.0-SNAPSHOT + + 4.0.0 + com.alibaba.otter + canal.client-adapter + pom + canal client adapter module for otter ${project.version} + + common + logger + hbase + + + \ No newline at end of file diff --git a/client-launcher/pom.xml b/client-launcher/pom.xml new file mode 100644 index 00000000..952aeb8a --- /dev/null +++ b/client-launcher/pom.xml @@ -0,0 +1,150 @@ + + + + canal + com.alibaba.otter + 1.1.0-SNAPSHOT + + 4.0.0 + com.alibaba.otter + canal.client-launcher + jar + canal client launcher module for otter ${project.version} + + + com.alibaba.otter + client-adapter.common + ${project.version} + + + + com.alibaba.otter + canal.client + ${project.version} + + + + com.alibaba.otter + canal.kafka.client + ${project.version} + + + org.yaml + snakeyaml + 1.17 + + + + + com.alibaba.otter + client-adapter.logger + ${project.version} + jar-with-dependencies + + + com.alibaba.otter + client-adapter.hbase + ${project.version} + jar-with-dependencies + + + + + + + + maven-jar-plugin + + + true + + + **/logback.xml + **/canal-client.yml + + + + + + org.apache.maven.plugins + maven-assembly-plugin + + 2.2.1 + + + assemble + + single + + package + + + + false + false + + + + + + + + dev + + true + + env + !release + + + + + + + maven-assembly-plugin + + + + ${basedir}/src/main/assembly/dev.xml + + canal_client + ${project.build.directory} + + + + + + + + + release + + + env + release + + + + + + + maven-assembly-plugin + + + + ${basedir}/src/main/assembly/release.xml + + + ${project.artifactId}-${project.version} + + ${project.parent.build.directory} + + + + + + + + \ No newline at end of file diff --git a/client-launcher/src/main/assembly/dev.xml b/client-launcher/src/main/assembly/dev.xml new file mode 100644 index 00000000..a3cbc20c --- /dev/null +++ b/client-launcher/src/main/assembly/dev.xml @@ -0,0 +1,57 @@ + + dist + + dir + + false + + + . + / + + README* + + + + ./src/main/bin + bin + + **/* + + 0755 + + + ./src/main/conf + /conf + + **/* + + + + ./src/main/resources + /conf + + **/* + + + META-INF/** + + + + target + logs + + **/* + + + + + + lib + + junit:junit + + + + diff --git a/client-launcher/src/main/assembly/release.xml b/client-launcher/src/main/assembly/release.xml new file mode 100644 index 00000000..9350da6e --- /dev/null +++ b/client-launcher/src/main/assembly/release.xml @@ -0,0 +1,57 @@ + + dist + + tar.gz + + false + + + . + / + + README* + + + + ./src/main/bin + bin + + **/* + + 0755 + + + ./src/main/conf + /conf + + **/* + + + + ./src/main/resources + /conf + + **/* + + + META-INF/** + + + + target + logs + + **/* + + + + + + lib + + junit:junit + + + + diff --git a/client-launcher/src/main/bin/startup.bat b/client-launcher/src/main/bin/startup.bat new file mode 100644 index 00000000..33fd7945 --- /dev/null +++ b/client-launcher/src/main/bin/startup.bat @@ -0,0 +1,25 @@ +@echo off +@if not "%ECHO%" == "" echo %ECHO% +@if "%OS%" == "Windows_NT" setlocal + +set ENV_PATH=.\ +if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0% + +set conf_dir=%ENV_PATH%\..\conf +set canal_conf=%conf_dir%\canal-client.yml +set logback_configurationFile=%conf_dir%\logback.xml + +set CLASSPATH=%conf_dir% +set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH% + +set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m +set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8 +set JAVA_DEBUG_OPT= -server -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=9099,server=y,suspend=n +set CANAL_OPTS= -DappName=otter-canal -Dlogback.configurationFile="%logback_configurationFile%" -Dcanal.conf="%canal_conf%" + +set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %JAVA_DEBUG_OPT% %CANAL_OPTS% + +set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.client.ClientLauncher +echo start cmd : %CMD_STR% + +java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.client.ClientLauncher \ No newline at end of file diff --git a/client-launcher/src/main/bin/startup.sh b/client-launcher/src/main/bin/startup.sh new file mode 100644 index 00000000..d467b4d2 --- /dev/null +++ b/client-launcher/src/main/bin/startup.sh @@ -0,0 +1,100 @@ +#!/bin/bash + +current_path=`pwd` +case "`uname`" in + Linux) + bin_abs_path=$(readlink -f $(dirname $0)) + ;; + *) + bin_abs_path=`cd $(dirname $0); pwd` + ;; +esac +base=${bin_abs_path}/.. +canal_conf=$base/conf/canal-client.yml +logback_configurationFile=$base/conf/logback.xml +export LANG=en_US.UTF-8 +export BASE=$base + +if [ -f $base/bin/canal_client.pid ] ; then + echo "found canal_client.pid , Please run stop.sh first ,then startup.sh" 2>&2 + exit 1 +fi + +## set java path +if [ -z "$JAVA" ] ; then + JAVA=$(which java) +fi + +ALIBABA_JAVA="/usr/alibaba/java/bin/java" +TAOBAO_JAVA="/opt/taobao/java/bin/java" +if [ -z "$JAVA" ]; then + if [ -f $ALIBABA_JAVA ] ; then + JAVA=$ALIBABA_JAVA + elif [ -f $TAOBAO_JAVA ] ; then + JAVA=$TAOBAO_JAVA + else + echo "Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH." 2>&2 + exit 1 + fi +fi + +case "$#" +in +0 ) + ;; +1 ) + var=$* + if [ -f $var ] ; then + canal_conf=$var + else + echo "THE PARAMETER IS NOT CORRECT.PLEASE CHECK AGAIN." + exit + fi;; +2 ) + var=$1 + if [ -f $var ] ; then + canal_conf=$var + else + if [ "$1" = "debug" ]; then + DEBUG_PORT=$2 + DEBUG_SUSPEND="n" + JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND" + fi + fi;; +* ) + echo "THE PARAMETERS MUST BE TWO OR LESS.PLEASE CHECK AGAIN." + exit;; +esac + +str=`file -L $JAVA | grep 64-bit` +if [ -n "$str" ]; then + JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -Xss256k -XX:-UseAdaptiveSizePolicy -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseFastAccessorMethods -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError" +else + JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m " +fi + +JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8" +CANAL_OPTS="-DappName=otter-canal -Dlogback.configurationFile=$logback_configurationFile -Dcanal.conf=$canal_conf" + +if [ -e $canal_conf -a -e $logback_configurationFile ] +then + + for i in $base/lib/*; + do CLASSPATH=$i:"$CLASSPATH"; + done + CLASSPATH="$base/conf:$CLASSPATH"; + + echo "cd to $bin_abs_path for workaround relative path" + cd $bin_abs_path + + echo LOG CONFIGURATION : $logback_configurationFile + echo canal conf : $canal_conf + echo CLASSPATH :$CLASSPATH + $JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.client.ClientLauncher 1>>$base/logs/canal_client.log 2>&1 & + echo $! > $base/bin/canal_client.pid + + echo "cd to $current_path for continue" + cd $current_path +else + echo "canal client conf("$canal_conf") OR log configration file($logback_configurationFile) is not exist,please create then first!" +fi diff --git a/client-launcher/src/main/bin/stop.sh b/client-launcher/src/main/bin/stop.sh new file mode 100644 index 00000000..0cfcc712 --- /dev/null +++ b/client-launcher/src/main/bin/stop.sh @@ -0,0 +1,65 @@ +#!/bin/bash + +cygwin=false; +linux=false; +case "`uname`" in + CYGWIN*) + cygwin=true + ;; + Linux*) + linux=true + ;; +esac + +get_pid() { + STR=$1 + PID=$2 + if $cygwin; then + JAVA_CMD="$JAVA_HOME\bin\java" + JAVA_CMD=`cygpath --path --unix $JAVA_CMD` + JAVA_PID=`ps |grep $JAVA_CMD |awk '{print $1}'` + else + if $linux; then + if [ ! -z "$PID" ]; then + JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` + else + JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'` + fi + else + if [ ! -z "$PID" ]; then + JAVA_PID=`ps aux |grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` + else + JAVA_PID=`ps aux |grep "$STR"|grep -v grep|awk '{print $2}'` + fi + fi + fi + echo $JAVA_PID; +} + +base=`dirname $0`/.. +pidfile=$base/bin/canal_client.pid +if [ ! -f "$pidfile" ];then + echo "canal client is not running. exists" + exit +fi + +pid=`cat $pidfile` +if [ "$pid" == "" ] ; then + pid=`get_pid "appName=otter-canal"` +fi + +echo -e "`hostname`: stopping canal $pid ... " +kill $pid + +LOOPS=0 +while (true); +do + gpid=`get_pid "appName=otter-canal" "$pid"` + if [ "$gpid" == "" ] ; then + echo "Oook! cost:$LOOPS" + `rm $pidfile` + break; + fi + let LOOPS=LOOPS+1 + sleep 1 +done \ No newline at end of file diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java new file mode 100644 index 00000000..c94fcd28 --- /dev/null +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java @@ -0,0 +1,65 @@ +package com.alibaba.otter.canal.client; + +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.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.yaml.snakeyaml.Yaml; + +import java.io.FileInputStream; + +public class ClientLauncher { + + private static final String CLASSPATH_URL_PREFIX = "classpath:"; + private static final Logger logger = LoggerFactory.getLogger(ClientLauncher.class); + + public static void main(String[] args) { + try { + logger.info("## set default uncaught exception handler"); + setGlobalUncaughtExceptionHandler(); + + logger.info("## load canal client configurations"); + String conf = System.getProperty("client.conf", "classpath:canal-client.yml"); + CanalClientConfig canalClientConfig; + if (conf.startsWith(CLASSPATH_URL_PREFIX)) { + conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); + canalClientConfig = new Yaml().loadAs(ClientLauncher.class.getClassLoader().getResourceAsStream(conf), + CanalClientConfig.class); + } else { + canalClientConfig = new Yaml().loadAs(new FileInputStream(conf), CanalClientConfig.class); + } + logger.info("## start the canal client adapters."); + final CanalAdapterLoader adapterLoader = new CanalAdapterLoader(canalClientConfig); + adapterLoader.init(); + logger.info("## the canal client adapters are running now ......"); + Runtime.getRuntime().addShutdownHook(new Thread() { + + public void run() { + try { + logger.info("## stop the canal client adapters"); + adapterLoader.destroy(); + } catch (Throwable e) { + logger.warn("## something goes wrong when stopping canal client adapters:", e); + } finally { + logger.info("## canal client adapters are down."); + } + } + + }); + } catch (Throwable e) { + logger.error("## something goes wrong when starting up the canal client adapters:", e); + System.exit(0); + } + } + + private static void setGlobalUncaughtExceptionHandler() { + Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + + @Override + public void uncaughtException(Thread t, Throwable e) { + logger.error("UnCaughtException", e); + } + }); + } +} diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java new file mode 100644 index 00000000..9dd328e3 --- /dev/null +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java @@ -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> 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> futures = new ArrayList<>(); + // 组间适配器并行运行 + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; + futures.add(groupInnerExecutorService.submit(new Callable() { + + @Override + public Boolean call() { + try { + // 组内适配器穿行运行，尽量不要配置组内适配器 + for (CanalOuterAdapter c : adapters) { + long begin = System.currentTimeMillis(); + 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 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(); +} diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java new file mode 100644 index 00000000..6b35e452 --- /dev/null +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java @@ -0,0 +1,168 @@ +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.KafkaCanalConnectors; +import com.alibaba.otter.canal.protocol.Message; +import org.apache.kafka.clients.consumer.CommitFailedException; +import org.apache.kafka.common.errors.WakeupException; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { + + private KafkaCanalConnector connector; + + private String topic; + + public CanalAdapterKafkaWorker(String zkServers, String bootstrapServers, String topic, String groupId, + List> canalOuterAdapters){ + this.canalOuterAdapters = canalOuterAdapters; + this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + this.topic = topic; + this.canalDestination = topic; + connector = KafkaCanalConnectors.newKafkaConnector(zkServers, bootstrapServers, topic, null, groupId); + // connector.setSessionTimeout(5L, TimeUnit.MINUTES); + + // super.initSwitcher(topic); + } + + @Override + public void start() { + if (!running) { + thread = new Thread(new Runnable() { + + @Override + public void run() { + process(); + } + }); + thread.setUncaughtExceptionHandler(handler); + running = true; + thread.start(); + } + } + + @Override + public void stop() { + try { + if (!running) { + return; + } + + connector.stopRunning(); + running = false; + + // if (switcher != null && !switcher.state()) { + // switcher.set(true); + // } + + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + groupInnerExecutorService.shutdown(); + logger.info("topic {} connectors' worker thread dead!", this.topic); + for (List outerAdapters : canalOuterAdapters) { + for (CanalOuterAdapter adapter : outerAdapters) { + adapter.destroy(); + } + } + logger.info("topic {} all connectors destroyed!", this.topic); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void process() { + while (!running) + ; + ExecutorService executor = Executors.newFixedThreadPool(1); + final AtomicBoolean executing = new AtomicBoolean(true); + while (running) { + try { + logger.info("=============> Start to connect topic: {} <=============", this.topic); + connector.connect(); + logger.info("=============> Start to subscribe topic: {}<=============", this.topic); + connector.subscribe(); + logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); + while (running) { + try { + // switcher.get(); //等待开关开启 + + final Message message = connector.getWithoutAck(); + + executing.set(true); + if (message != null) { + executor.submit(new Runnable() { + + @Override + public void run() { + try { + if (logger.isDebugEnabled()) { + logger.debug("topic: {} batchId: {} batchSize: {} ", + topic, + message.getId(), + message.getEntries().size()); + } + long begin = System.currentTimeMillis(); + writeOut(message); + long now = System.currentTimeMillis(); + if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { + logger.error("topic: {} batchId {} elapsed time: {} ms", + topic, + message.getId(), + now - begin); + } + 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 + connector.ack(); + Thread.sleep(500); + } + } else { + connector.ack(); + } + } catch (CommitFailedException e) { + logger.warn(e.getMessage()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + TimeUnit.SECONDS.sleep(1L); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + executor.shutdown(); + + try { + connector.unsubscribe(); + } catch (WakeupException e) { + // No-op. Continue process + } + connector.disconnnect(); + logger.info("=============> Disconnect topic: {} <=============", this.topic); + } +} diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java new file mode 100644 index 00000000..b0fe35c5 --- /dev/null +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java @@ -0,0 +1,164 @@ +package com.alibaba.otter.canal.client.adapter.loader; + +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * 外部适配器的加载器 + * + * @author machengyuan 2018-8-19 下午11:45:49 + * @version 1.0.0 + */ +public class CanalAdapterLoader { + + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); + + private CanalClientConfig canalClientConfig; + + private Map canalWorkers = new HashMap<>(); + + private Map canalKafkaWorkers = new HashMap<>(); + + private ExtensionLoader loader; + + public CanalAdapterLoader(CanalClientConfig canalClientConfig){ + this.canalClientConfig = canalClientConfig; + } + + /** + * 初始化canal-client、 canal-client-kafka的适配器 + */ + public void init() { + // canal instances 和 kafka topics 配置不能同时为空 + if (canalClientConfig.getCanalInstances().isEmpty() && canalClientConfig.getKafkaTopics().isEmpty()) { + throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics"); + } + + loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, + "" /* TODO canalClientConfig.getClassloaderPolicy() */); + + String canalServerHost = this.canalClientConfig.getCanalServerHost(); + SocketAddress sa = null; + if (canalServerHost != null) { + String[] ipPort = canalServerHost.split(":"); + sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1])); + } + String zkHosts = this.canalClientConfig.getZookeeperHosts(); + + if (zkHosts == null && sa == null) { + throw new RuntimeException("Blank config property: canalServerHost or zookeeperHosts"); + } + + // 初始化canal-client的适配器 + for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { + List> canalOuterAdapterGroups = new ArrayList<>(); + + for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { + List canalOutConnectors = new ArrayList<>(); + for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) { + loadConnector(c, canalOutConnectors); + } + canalOuterAdapterGroups.add(canalOutConnectors); + } + CanalAdapterWorker worker; + if (zkHosts != null) { + worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups); + } else { + worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); + } + canalWorkers.put(instance.getInstance(), worker); + worker.start(); + logger.info("Start adapter for canal instance: {} succeed", instance.getInstance()); + } + + // 初始化canal-client-kafka的适配器 + for (CanalClientConfig.KafkaTopic kafkaTopic : canalClientConfig.getKafkaTopics()) { + for (CanalClientConfig.Group group : kafkaTopic.getGroups()) { + List> canalOuterAdapterGroups = new ArrayList<>(); + + List canalOuterAdapters = new ArrayList<>(); + + for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) { + // for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) { + loadConnector(config, canalOuterAdapters); + // } + } + canalOuterAdapterGroups.add(canalOuterAdapters); + + String zkServers = canalClientConfig.getZookeeperHosts(); + CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkServers, + canalClientConfig.getBootstrapServers(), + kafkaTopic.getTopic(), + group.getGroupId(), + canalOuterAdapterGroups); + canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker); + canalKafkaWorker.start(); + logger.info("Start adapter for canal-client kafka topic: {} succeed", + kafkaTopic.getTopic() + "-" + group.getGroupId()); + } + } + } + + private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { + try { + CanalOuterAdapter adapter = loader.getExtension(config.getName()); + ClassLoader cl = Thread.currentThread().getContextClassLoader(); + // 替换ClassLoader + Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); + adapter.init(config); + Thread.currentThread().setContextClassLoader(cl); + canalOutConnectors.add(adapter); + logger.info("Load canal adapter: {} succeed", config.getName()); + } catch (Exception e) { + logger.error("Load canal adapter: {} failed", config.getName(), e); + } + } + + /** + * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁 + */ + public void destroy() { + if (canalWorkers.size() > 0) { + ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); + for (CanalAdapterWorker v : canalWorkers.values()) { + final CanalAdapterWorker caw = v; + stopExecutorService.submit(new Runnable() { + + @Override + public void run() { + caw.stop(); + } + }); + } + stopExecutorService.shutdown(); + } + if (canalKafkaWorkers.size() > 0) { + ExecutorService stopKafkaExecutorService = Executors.newFixedThreadPool(canalKafkaWorkers.size()); + for (CanalAdapterKafkaWorker v : canalKafkaWorkers.values()) { + final CanalAdapterKafkaWorker cakw = v; + stopKafkaExecutorService.submit(new Runnable() { + + @Override + public void run() { + cakw.stop(); + } + }); + } + stopKafkaExecutorService.shutdown(); + } + logger.info("All canal adapters destroyed"); + } +} diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java new file mode 100644 index 00000000..bca40041 --- /dev/null +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java @@ -0,0 +1,187 @@ +package com.alibaba.otter.canal.client.adapter.loader; + +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.client.CanalConnectors; +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker; +import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; +import com.alibaba.otter.canal.protocol.Message; + +import java.net.SocketAddress; +import java.util.List; +import java.util.concurrent.Executors; + +/** + * 原生canal-server对应的client适配器工作线程 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class CanalAdapterWorker extends AbstractCanalAdapterWorker { + + private static final int BATCH_SIZE = 50; + private static final int SO_TIMEOUT = 0; + + private CanalConnector connector; + + /** + * 单台client适配器worker的构造方法 + * + * @param canalDestination canal实例名 + * @param address canal-server地址 + * @param canalOuterAdapters 外部适配器组 + */ + public CanalAdapterWorker(String canalDestination, SocketAddress address, + List> canalOuterAdapters){ + this.canalOuterAdapters = canalOuterAdapters; + this.canalDestination = canalDestination; + groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + connector = CanalConnectors.newSingleConnector(address, canalDestination, "", ""); + } + + /** + * HA模式下client适配器worker的构造方法 + * + * @param canalDestination canal实例名 + * @param zookeeperHosts zookeeper地址 + * @param canalOuterAdapters 外部适配器组 + */ + public CanalAdapterWorker(String canalDestination, String zookeeperHosts, + List> canalOuterAdapters){ + this.canalOuterAdapters = canalOuterAdapters; + this.canalDestination = canalDestination; + groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", ""); + ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT); + + // super.initSwitcher(canalDestination); + } + + @Override + public void start() { + if (!running) { + thread = new Thread(new Runnable() { + + @Override + public void run() { + process(); + } + }); + thread.setUncaughtExceptionHandler(handler); + thread.start(); + running = true; + } + } + + @Override + public void stop() { + try { + if (!running) { + return; + } + + // if (switcher != null && !switcher.state()) { + // switcher.set(true); + // } + + connector.stopRunning(); + running = false; + + logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination); + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + groupInnerExecutorService.shutdown(); + logger.info("destination {} adapters' worker thread dead!", canalDestination); + for (List outerAdapters : canalOuterAdapters) { + for (CanalOuterAdapter adapter : outerAdapters) { + adapter.destroy(); + } + } + logger.info("destination {} all adapters destroyed!", canalDestination); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void process() { + while (!running) + ; // waiting until running == true + while (running) { + try { + // if (switcher != null) { + // switcher.get(); + // } + logger.info("=============> Start to connect destination: {} <=============", this.canalDestination); + connector.connect(); + logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination); + connector.subscribe(); + logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination); + while (running) { + // try { + // if (switcher != null) { + // switcher.get(); + // } + // } catch (TimeoutException e) { + // break; + // } + + // server配置canal.instance.network.soTimeout(默认: 30s) + // 范围内未与server交互，server将关闭本次socket连接 + Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据 + long batchId = message.getId(); + try { + int size = message.getEntries().size(); + + if (batchId == -1 || size == 0) { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + // ignore + } + } else { + if (logger.isDebugEnabled()) { + logger.debug("destination: {} batchId: {} batchSize: {} ", + this.canalDestination, + batchId, + size); + } + long begin = System.currentTimeMillis(); + writeOut(message); + long now = System.currentTimeMillis(); + if (logger.isDebugEnabled()) { + logger.debug("destination: {} batchId: {} elapsed time: {} ms", + this.canalDestination, + batchId, + now - begin); + } + } + connector.ack(batchId); // 提交确认 + } catch (Exception e) { + connector.rollback(batchId); // 处理失败, 回滚数据 + throw e; + } + } + + } catch (Exception e) { + logger.error("process error!", e); + } finally { + connector.disconnect(); + logger.info("=============> Disconnect destination: {} <=============", this.canalDestination); + } + + if (running) { // is reconnect + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + // ignore + } + } + + } + } +} diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml new file mode 100644 index 00000000..2412e38d --- /dev/null +++ b/client-launcher/src/main/resources/canal-client.yml @@ -0,0 +1,22 @@ +canalServerHost: 127.0.0.1:11111 +#zookeeperHosts: 127.0.0.1:2181 +#bootstrapServers: kafka1.mytest.com:9092,kafka2.mytest.com:9092 + +canalInstances: +- instance: example + adapterGroups: + - outAdapters: + - name: logger + - name: hbase + hosts: slave1:2181 + properties: {znodeParent: "/hbase-unsecure"} + +#kafkaTopics: +#- topic: devmysql4308 +# groups: +# - groupId: devmysql4308_es +# adapters: +# - name: es +# hosts: +# zkHosts: +# properties: {clusterName: es-service-test} diff --git a/client-launcher/src/main/resources/logback.xml b/client-launcher/src/main/resources/logback.xml new file mode 100644 index 00000000..665954f7 --- /dev/null +++ b/client-launcher/src/main/resources/logback.xml @@ -0,0 +1,47 @@ + + + + + %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n + + + + + + + destination + canal_client + + + + ../logs/${destination}.log + + + ../logs/%d{yyyy-MM-dd}/${destination}-%d{yyyy-MM-dd}-%i.log.gz + + + 512MB + + 60 + + + + %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/pom.xml b/pom.xml index 66b69ca2..3fc16eab 100644 --- a/pom.xml +++ b/pom.xml @@ -120,6 +120,8 @@ kafka kafka-client prometheus + client-adapter + client-launcher @@ -332,7 +334,7 @@ org.apache.maven.plugins maven-compiler-plugin - 3.8.0 + 3.7.0 ${java_source_version} ${java_target_version}