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 index 24d43a0f..cf69fde5 100644 --- 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 @@ -1,6 +1,9 @@ package com.alibaba.otter.canal.client.adapter.support; -import java.io.*; +import java.io.BufferedReader; +import java.io.File; +import java.io.IOException; +import java.io.InputStreamReader; import java.net.MalformedURLException; import java.net.URL; import java.net.URLClassLoader; @@ -83,47 +86,6 @@ public class ExtensionLoader { 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())); - } - /** * 返回指定名字的扩展 * @@ -165,87 +127,6 @@ public class ExtensionLoader { 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); @@ -267,15 +148,6 @@ public class ExtensionLoader { } } - 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) { @@ -334,23 +206,18 @@ public class ExtensionLoader { // 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"); + externalLibDir = new File(File.separator + this.getJarDirectoryPath() + File.separator + "canal-adapter" + + File.separator + "lib"); } + logger.info("extension classpath dir: " + externalLibDir.getAbsolutePath()); if (externalLibDir.exists()) { - File[] files = externalLibDir.listFiles(new FilenameFilter() { - - @Override - public boolean accept(File dir, String name) { - return name.endsWith(".jar"); - } - }); + File[] files = externalLibDir.listFiles((dir1, name) -> name.endsWith(".jar")); if (files != null) { for (File f : files) { - URL url = null; + URL url; try { url = f.toURI().toURL(); } catch (MalformedURLException e) { diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java index cb1cc274..d9b09161 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java @@ -99,11 +99,11 @@ public class ESSyncConfigLoader { InputStream in = null; try { // 先取本地文件，再取类路径 - File configFile = new File("config/" + config); + File configFile = new File("../config/" + config); if (configFile.exists()) { in = new FileInputStream(configFile); } else { - in = ESSyncConfigLoader.class.getClassLoader().getResourceAsStream(config); + in = ESSyncConfigLoader.class.getClassLoader().getResourceAsStream(config); } if (in == null) { throw new RuntimeException("Config file: " + config + " not found."); diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index 3927c6ab..272ef1a9 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -9,6 +9,7 @@ import javax.sql.DataSource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.fastjson.JSON; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; @@ -38,9 +39,6 @@ public class ESSyncService { } public void sync(Dml dml) { - if (logger.isDebugEnabled()) { - logger.debug("DML: {}", dml.toString()); - } long begin = System.currentTimeMillis(); String database = dml.getDatabase(); String table = dml.getTable(); @@ -74,6 +72,9 @@ public class ESSyncService { esSyncConfigs.size(), dml.getDestination()); } + if (logger.isDebugEnabled()) { + logger.debug("DML: {}", JSON.toJSONString(dml)); + } } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java index cd5407d8..5b6cf2bd 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java @@ -124,7 +124,7 @@ public class MappingConfigLoader { InputStream in = null; try { // 先取本地文件，再取类路径 - File configFile = new File("config/" + config); + File configFile = new File("../config/" + config); if (configFile.exists()) { in = new FileInputStream(configFile); } else { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java index 9606ee97..420640af 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java @@ -6,6 +6,7 @@ import org.apache.hadoop.hbase.util.Bytes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.fastjson.JSON; 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; @@ -38,7 +39,7 @@ public class HbaseSyncService { delete(config, dml); } if (logger.isDebugEnabled()) { - logger.debug("DML: {}", dml.toString()); + logger.debug("DML: {}", JSON.toJSONString(dml)); } } } catch (Exception e) { diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index c24674bd..6b5f71dd 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -47,7 +47,7 @@ org.springframework.boot spring-boot-configuration-processor - true + provided org.apache.curator @@ -79,7 +79,7 @@ jar-with-dependencies - true + provided com.alibaba.otter @@ -92,7 +92,7 @@ jar-with-dependencies - true + provided com.alibaba.otter @@ -105,72 +105,12 @@ jar-with-dependencies - true + provided - canal-adapter-launcher - - org.springframework.boot - spring-boot-maven-plugin - 2.0.1.RELEASE - - - - com.alibaba.otter - client-adapter.logger - - - com.alibaba.otter - client-adapter.hbase - - - com.alibaba.otter - client-adapter.elasticsearch - - - - - - - repackage - - - - - - maven-antrun-plugin - - - package - - run - - - - - - - - - - - - - - - - - - - - - - - org.apache.maven.plugins maven-dependency-plugin @@ -184,15 +124,81 @@ jar-with-dependencies - ${project.basedir}/target/lib + ${project.basedir}/target/canal-adapter/lib + + org.apache.maven.plugins + maven-assembly-plugin + 2.2.1 + + + assemble + + single + + package + + + + false + false + + + + org.apache.maven.plugins + maven-jar-plugin + 3.0.2 + + + + true + lib/ + com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication + + + + **/*.properties + **/*.xml + **/*.yml + static/** + templates/** + + + + + dev + + true + + env + !release + + + + + + + maven-assembly-plugin + + + ${basedir}/src/main/assembly/dev.xml + + canal-adapter + ${project.build.directory} + + + + + + + release @@ -207,32 +213,14 @@ maven-assembly-plugin - false ${basedir}/src/main/assembly/release.xml ${project.artifactId}-${project.version} ${project.basedir}/../../target - - - make-assembly - package - - single - - - - - - src/main/resources - - application.yml - - - diff --git a/client-adapter/launcher/src/main/assembly/dev.xml b/client-adapter/launcher/src/main/assembly/dev.xml new file mode 100644 index 00000000..1be787b5 --- /dev/null +++ b/client-adapter/launcher/src/main/assembly/dev.xml @@ -0,0 +1,62 @@ + + dist + + dir + + false + + + . + / + + README* + + + + ./src/main/bin + bin + + **/* + + 0755 + + + ./src/main/resources + /config + + **/* + + + + + ../elasticsearch/src/main/resources/es + /config/es + + **/* + + + + ../hbase/src/main/resources/hbase + /config/hbase + + **/* + + + + target + logs + + **/* + + + + + + lib + + junit:junit + + + + diff --git a/client-adapter/launcher/src/main/assembly/release.xml b/client-adapter/launcher/src/main/assembly/release.xml index 7bb12314..57a815c8 100644 --- a/client-adapter/launcher/src/main/assembly/release.xml +++ b/client-adapter/launcher/src/main/assembly/release.xml @@ -1,34 +1,71 @@ - canal-adapter + dist tar.gz + false - ${project.build.directory} + . / - *.jar + README* + + ./src/main/bin + bin + + **/* + + 0755 + + + ./src/main/resources + /config + + **/* + + + + + ../elasticsearch/src/main/resources/es + /config/es + + **/* + + + + ../hbase/src/main/resources/hbase + /config/hbase + + **/* + + + + target + logs + + **/* + + - ${project.build.directory}/lib + ${project.basedir}/target/canal-adapter/lib /lib/ - *.jar - - - - - ${project.build.directory}/config - /config/ - - ** + *-jar-with-dependencies.jar - / + + + lib + + junit:junit + + + \ No newline at end of file diff --git a/client-adapter/launcher/src/main/bin/restart.sh b/client-adapter/launcher/src/main/bin/restart.sh new file mode 100644 index 00000000..63749ce5 --- /dev/null +++ b/client-adapter/launcher/src/main/bin/restart.sh @@ -0,0 +1,5 @@ +#!/bin/bash + +sh stop.sh + +sh startup.sh diff --git a/client-adapter/launcher/src/main/bin/startup.bat b/client-adapter/launcher/src/main/bin/startup.bat new file mode 100755 index 00000000..00555432 --- /dev/null +++ b/client-adapter/launcher/src/main/bin/startup.bat @@ -0,0 +1,22 @@ +@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%\..\config + +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 ADAPTER_OPTS= -DappName=canal-adapter + +set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %ADAPTER_OPTS% + +set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication +echo start cmd : %CMD_STR% + +java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication \ No newline at end of file diff --git a/client-adapter/launcher/src/main/bin/startup.sh b/client-adapter/launcher/src/main/bin/startup.sh new file mode 100644 index 00000000..001f7fb9 --- /dev/null +++ b/client-adapter/launcher/src/main/bin/startup.sh @@ -0,0 +1,68 @@ +#!/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/config/canal.properties +#logback_configurationFile=$base/conf/logback.xml +export LANG=en_US.UTF-8 +export BASE=$base + +if [ -f $base/bin/adapter.pid ] ; then + echo "found adapter.pid , Please run stop.sh first ,then startup.sh" 2>&2 + exit 1 +fi + +if [ ! -d $base/logs ] ; then + mkdir -p $base/logs +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 + +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" +ADAPTER_OPTS="-DappName=canal-adapter" + +for i in $base/lib/*; + do CLASSPATH=$i:"$CLASSPATH"; +done +CLASSPATH="$base/config:$CLASSPATH"; + +echo "cd to $bin_abs_path for workaround relative path" +cd $bin_abs_path + +echo CLASSPATH :$CLASSPATH +$JAVA $JAVA_OPTS $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>$base/logs/adapter.log 2>&1 & +echo $! > $base/bin/adapter.pid + +echo "cd to $current_path for continue" +cd $current_path \ No newline at end of file diff --git a/client-adapter/launcher/src/main/bin/stop.sh b/client-adapter/launcher/src/main/bin/stop.sh new file mode 100644 index 00000000..fbdd5317 --- /dev/null +++ b/client-adapter/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/adapter.pid +if [ ! -f "$pidfile" ];then + echo "canal-adapter is not running. exists" + exit +fi + +pid=`cat $pidfile` +if [ "$pid" == "" ] ; then + pid=`get_pid "appName=canal-adapter"` +fi + +echo -e "`hostname`: stopping canal $pid ... " +kill $pid + +LOOPS=0 +while (true); +do + gpid=`get_pid "appName=canal-adapter" "$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-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index f0b4c6a2..894d123e 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -3,7 +3,7 @@ server: logging: level: com.alibaba.otter.canal.client.adapter.hbase: DEBUG - com.alibaba.otter.canal.client.adapter.es: TRACE + com.alibaba.otter.canal.client.adapter.es: DEBUG spring: jackson: date-format: yyyy-MM-dd HH:mm:ss diff --git a/example/src/main/java/com/alibaba/otter/canal/example/db/CanalConnectorClient.java b/example/src/main/java/com/alibaba/otter/canal/example/db/CanalConnectorClient.java index 844f60db..247f9e1f 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/db/CanalConnectorClient.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/db/CanalConnectorClient.java @@ -359,7 +359,7 @@ public abstract class CanalConnectorClient extends AbstractCanalLifeCycle implem logger.warn("parse delete index event, table: {}, sql: {}", table, sql); return; default: - logger.warn("parse unknown event: {}, table: {}, sql: {}", new String[]{eventType.name(), table, sql}); + logger.warn("parse unknown event: {}, table: {}, sql: {}", eventType.name(), table, sql); break; } } @@ -375,7 +375,7 @@ public abstract class CanalConnectorClient extends AbstractCanalLifeCycle implem public void whenOthers(CanalEntry.Header header, String sql) { String schema = header.getSchemaName(); String table = header.getTableName(); - logger.error("ignore event,schema: {},table: {},SQL: {}", new String[]{schema, table, sql}); + logger.error("ignore event,schema: {},table: {},SQL: {}", schema, table, sql); } public enum ExceptionStrategy {