diff --git a/client-adapter/example/pom.xml b/client-adapter/example/pom.xml deleted file mode 100644 index bf68d731..00000000 --- a/client-adapter/example/pom.xml +++ /dev/null @@ -1,149 +0,0 @@ - - - - canal.client-adapter - com.alibaba.otter - 1.1.1-SNAPSHOT - - 4.0.0 - com.alibaba.otter - client-adapter.example - jar - canal client adapter example module for otter ${project.version} - - - - com.alibaba.otter - client-adapter.common - ${project.version} - - - com.alibaba.otter - canal.client - ${canal_version} - - - - org.apache.rocketmq - rocketmq-client - 4.3.0 - - - - org.apache.kafka - kafka-clients - 1.1.1 - - - org.yaml - snakeyaml - 1.17 - - - - - com.alibaba.otter - client-adapter.logger - ${project.version} - - - - - - - - maven-jar-plugin - 3.0.2 - - - 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} - - - - - - - diff --git a/client-adapter/example/src/main/assembly/dev.xml b/client-adapter/example/src/main/assembly/dev.xml deleted file mode 100644 index a3cbc20c..00000000 --- a/client-adapter/example/src/main/assembly/dev.xml +++ /dev/null @@ -1,57 +0,0 @@ - - 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-adapter/example/src/main/assembly/release.xml b/client-adapter/example/src/main/assembly/release.xml deleted file mode 100644 index 9350da6e..00000000 --- a/client-adapter/example/src/main/assembly/release.xml +++ /dev/null @@ -1,57 +0,0 @@ - - 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-adapter/example/src/main/bin/startup.bat b/client-adapter/example/src/main/bin/startup.bat deleted file mode 100644 index 33fd7945..00000000 --- a/client-adapter/example/src/main/bin/startup.bat +++ /dev/null @@ -1,25 +0,0 @@ -@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-adapter/example/src/main/bin/startup.sh b/client-adapter/example/src/main/bin/startup.sh deleted file mode 100644 index 0e09ddff..00000000 --- a/client-adapter/example/src/main/bin/startup.sh +++ /dev/null @@ -1,100 +0,0 @@ -#!/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-client -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-adapter/example/src/main/bin/stop.sh b/client-adapter/example/src/main/bin/stop.sh deleted file mode 100644 index c96f2807..00000000 --- a/client-adapter/example/src/main/bin/stop.sh +++ /dev/null @@ -1,65 +0,0 @@ -#!/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-client"` -fi - -echo -e "`hostname`: stopping canal $pid ... " -kill $pid - -LOOPS=0 -while (true); -do - gpid=`get_pid "appName=otter-canal-client" "$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/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java deleted file mode 100644 index a705a526..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java +++ /dev/null @@ -1,66 +0,0 @@ -package com.alibaba.otter.canal.client; - -import java.io.FileInputStream; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.yaml.snakeyaml.Yaml; - -import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; - -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-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java deleted file mode 100644 index bbfd462d..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java +++ /dev/null @@ -1,168 +0,0 @@ -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 org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.client.adapter.support.MessageUtil; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.protocol.Message; - -/** - * 适配器工作线程抽象类 - * - * @author 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 (final OuterAdapter c : adapters) { - long begin = System.currentTimeMillis(); - MessageUtil.parse4Dml(canalDestination, message, new MessageUtil.Consumer() { - - @Override - public void accept(Dml dml) { - c.sync(dml); - } - }); - - 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) { - // ignore - } - } - } - } - - protected void writeOut(final FlatMessage flatMessage) { - List> futures = new ArrayList<>(); - // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; - futures.add(groupInnerExecutorService.submit(new Callable() { - - @Override - public Boolean call() { - try { - // 组内适配器穿行运行，尽量不要配置组内适配器 - for (OuterAdapter c : adapters) { - long begin = System.currentTimeMillis(); - Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage); - c.sync(dml); - 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) { - // ignore - } - } - } - } - - protected void writeOut(Message message, String topic) { - 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); - } - } - - protected void stopOutAdapters() { - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore - } - } - groupInnerExecutorService.shutdown(); - logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); - } - } - logger.info("topic all connectors destroyed!"); - } - - public abstract void start(); - - public abstract void stop(); -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java deleted file mode 100644 index b2d3e583..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ /dev/null @@ -1,178 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; - -import org.apache.kafka.clients.consumer.CommitFailedException; -import org.apache.kafka.common.errors.WakeupException; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; -import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.protocol.Message; - -/** - * kafka对应的client适配器工作线程 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @version 1.0.0 - */ -public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { - - private KafkaCanalConnector connector; - - private String topic; - - private boolean flatMessage; - - public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, - List> canalOuterAdapters, boolean flatMessage){ - this.canalOuterAdapters = canalOuterAdapters; - this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); - this.topic = topic; - this.canalDestination = topic; - this.flatMessage = flatMessage; - connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage); - // connector.setSessionTimeout(1L, TimeUnit.MINUTES); - - // 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 (OuterAdapter 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.newSingleThreadExecutor(); - // 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(); //等待开关开启 - - List messages; - if (!flatMessage) { - messages = connector.getWithoutAck(); - } else { - messages = connector.getFlatMessageWithoutAck(100L, TimeUnit.MILLISECONDS); - } - if (messages != null) { - for (final Object message : messages) { - if (message instanceof FlatMessage) { - writeOut((FlatMessage) message); - } else { - writeOut((Message) message); - } - // executing.set(true); - // if (message != null) { - // executor.submit(new Runnable() { - // - // @Override - // public void run() { - // try { - // if (message instanceof FlatMessage) { - // writeOut((FlatMessage) message); - // } else { - // writeOut((Message) message); - // } - // } catch (Exception e) { - // logger.error(e.getMessage(), e); - // } finally { - // executing.compareAndSet(true, false); - // } - // } - // }); - // - // // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端 - // long currentTS = System.currentTimeMillis(); - // while (executing.get()) { - // // 大于10秒未消费完ack一次keep alive - // if (System.currentTimeMillis() - currentTS > 10000) { - // connector.ack(); - // currentTS = System.currentTimeMillis(); - // } - // } - // } - } - } - connector.ack(); - } catch (CommitFailedException e) { - logger.warn(e.getMessage()); - } catch (Exception e) { - logger.error(e.getMessage(), e); - TimeUnit.SECONDS.sleep(1L); - } - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - executor.shutdown(); - - try { - connector.unsubscribe(); - } catch (WakeupException e) { - // No-op. Continue process - } - connector.disconnect(); - logger.info("=============> Disconnect topic: {} <=============", this.topic); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java deleted file mode 100644 index 51b12a0a..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ /dev/null @@ -1,177 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -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; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; -import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; -import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; - -/** - * MQ外部适配器的加载器 - * - * @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 canalMQWorker = new HashMap<>(); - - private ExtensionLoader loader; - - public CanalAdapterLoader(CanalClientConfig canalClientConfig){ - this.canalClientConfig = canalClientConfig; - } - - /** - * 初始化canal-client、 canal-client-rocketmq的适配器 - */ - public void init() { - // canal instances 和 mq topics 配置不能同时为空 - if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) { - throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); - } - - loader = ExtensionLoader.getExtensionLoader(OuterAdapter.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(); - - // 初始化canal-client的适配器 - if (canalClientConfig.getCanalInstances() != null) { - for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - - for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { - List canalOutConnectors = new ArrayList<>(); - for (OuterAdapterConfig 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-mq的适配器 - if (canalClientConfig.getMqTopics() != null) { - for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { - for (CanalClientConfig.Group group : topic.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - - List canalOuterAdapters = new ArrayList<>(); - - for (OuterAdapterConfig config : group.getOutAdapters()) { - loadConnector(config, canalOuterAdapters); - } - canalOuterAdapterGroups.add(canalOuterAdapters); - if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) { - CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig.getBootstrapServers(), - topic.getTopic(), - group.getGroupId(), - canalOuterAdapterGroups); - canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); - rocketMQWorker.start(); - } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) { - CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig.getBootstrapServers(), - topic.getTopic(), - group.getGroupId(), - canalOuterAdapterGroups, - canalClientConfig.getFlatMessage()); - canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker); - canalKafkaWorker.start(); - } - logger.info("Start adapter for canal-client rocketmq topic: {} succeed", topic.getTopic() + "-" - + group.getGroupId()); - - } - } - } - } - - private void loadConnector(OuterAdapterConfig config, List canalOutConnectors) { - try { - OuterAdapter 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 (canalMQWorker.size() > 0) { - ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size()); - for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) { - final AbstractCanalAdapterWorker worker = tmp; - stopMQWokerService.submit(new Runnable() { - - @Override - public void run() { - worker.stop(); - } - }); - } - stopMQWokerService.shutdown(); - } - logger.info("All canal adapters destroyed"); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java deleted file mode 100644 index 9c126297..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java +++ /dev/null @@ -1,124 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; - -import org.apache.kafka.clients.consumer.CommitFailedException; -import org.apache.kafka.common.errors.WakeupException; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; -import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; -import com.alibaba.otter.canal.protocol.Message; - -/** - * kafka对应的client适配器工作线程 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @version 1.0.0 - */ -public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { - - private RocketMQCanalConnector connector; - - private String topic; - - public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, - List> canalOuterAdapters){ - logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); - this.canalOuterAdapters = canalOuterAdapters; - this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); - this.topic = topic; - this.canalDestination = topic; - connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId); - } - - @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; - logger.info("Stop topic {} out adapters begin", this.topic); - stopOutAdapters(); - logger.info("Stop topic {} out adapters end", this.topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { - while (!running) - ; - ExecutorService executor = Executors.newSingleThreadExecutor(); - while (running) { - try { - logger.info("=============> Start to connect topic: {} <=============", this.topic); - connector.connect(); - logger.info("=============> Start to subscribe topic: {}<=============", this.topic); - connector.subscribe(); - logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); - while (running) { - try { - // switcher.get(); //等待开关开启 - - final Message message = connector.getWithoutAck(1); - if (message != null) { - executor.submit(new Runnable() { - - @Override - public void run() { - try { - writeOut(message, topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - connector.ack(message.getId()); - } - }); - } else { - logger.debug("Message is null"); - } - } 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.stopRunning(); - logger.info("=============> Disconnect topic: {} <=============", this.topic); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java deleted file mode 100644 index e72ed2d5..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java +++ /dev/null @@ -1,186 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.net.SocketAddress; -import java.util.List; -import java.util.concurrent.Executors; - -import com.alibaba.otter.canal.client.CanalConnector; -import com.alibaba.otter.canal.client.CanalConnectors; -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; -import com.alibaba.otter.canal.protocol.Message; - -/** - * 原生canal-server对应的client适配器工作线程 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @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 (OuterAdapter 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-adapter/example/src/main/resources/canal-client.yml b/client-adapter/example/src/main/resources/canal-client.yml deleted file mode 100644 index de24e756..00000000 --- a/client-adapter/example/src/main/resources/canal-client.yml +++ /dev/null @@ -1,18 +0,0 @@ -canalServerHost: 127.0.0.1:11111 -#zookeeperHosts: slave1:2181 -#bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876 -flatMessage: false - -canalInstances: -- instance: example - adapterGroups: - - outAdapters: - - name: logger - -#mqTopics: -#- mqMode: rocketmq -# topic: example -# groups: -# - groupId: example -# outAdapters: -# - name: logger diff --git a/client-adapter/example/src/main/resources/logback.xml b/client-adapter/example/src/main/resources/logback.xml deleted file mode 100644 index e74c84f0..00000000 --- a/client-adapter/example/src/main/resources/logback.xml +++ /dev/null @@ -1,47 +0,0 @@ - - - - - %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/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index a48d7401..f48cdd10 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -43,48 +43,60 @@ - - - - org.apache.maven.plugins - maven-assembly-plugin - 2.4 - - - jar-with-dependencies - - - - - make-assembly - package - - single - - - - - - maven-antrun-plugin - - - package - - run - - - - - - - - - - - - - - - + + + release + + + env + release + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + + + + + diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index cb8aee72..f10dfaac 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -104,6 +104,18 @@ org.springframework.boot spring-boot-maven-plugin 2.0.1.RELEASE + + + + com.alibaba.otter + client-adapter.hbase + + + com.alibaba.otter + client-adapter.logger + + + @@ -150,24 +162,43 @@ - - maven-assembly-plugin - - false - - assembly.xml - - - - - make-assembly - package - - single - - - - + + + + release + + + env + release + + + + + + + maven-assembly-plugin + + false + + ${basedir}/src/main/assembly/release.xml + + ${project.artifactId}-${project.version} + ${project.basedir}/../../target + + + + make-assembly + package + + single + + + + + + + + diff --git a/client-adapter/launcher/assembly.xml b/client-adapter/launcher/src/main/assembly/release.xml similarity index 100% rename from client-adapter/launcher/assembly.xml rename to client-adapter/launcher/src/main/assembly/release.xml diff --git a/client-adapter/logger/pom.xml b/client-adapter/logger/pom.xml index 8f2b4edb..e9a146aa 100644 --- a/client-adapter/logger/pom.xml +++ b/client-adapter/logger/pom.xml @@ -21,27 +21,39 @@ - - - - org.apache.maven.plugins - maven-assembly-plugin - 2.4 - - - jar-with-dependencies - - - - - make-assembly - package - - single - - - - - - + + + release + + + env + release + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + + diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index 7ca65de1..aafe6395 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -24,7 +24,6 @@ logger hbase launcher - example