diff --git a/deployer/pom.xml b/deployer/pom.xml index 2de1ce1e..76777f03 100644 --- a/deployer/pom.xml +++ b/deployer/pom.xml @@ -40,6 +40,7 @@ **/canal.properties **/spring/** **/example/** + **/kafka.yml diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index b328feb7..1e51990e 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -22,6 +22,7 @@ public class CanalConstants { public static final String CANAL_AUTO_SCAN = ROOT + "." + "auto.scan"; public static final String CANAL_AUTO_SCAN_INTERVAL = ROOT + "." + "auto.scan.interval"; public static final String CANAL_CONF_DIR = ROOT + "." + "conf.dir"; + public static final String CANAL_SERVER_MODE = ROOT + "." + "serverMode"; public static final String CANAL_DESTINATION_SPLIT = ","; public static final String GLOBAL_NAME = "global"; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 828f49e2..2524186a 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -3,6 +3,8 @@ package com.alibaba.otter.canal.deployer; import java.io.FileInputStream; import java.util.Properties; +import com.alibaba.otter.canal.kafka.CanalKafkaStarter; +import com.alibaba.otter.canal.server.CanalServerStarter; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,6 +53,17 @@ public class CanalLauncher { } }); + + CanalServerStarter canalServerStarter = null; + String serverMode = properties.getProperty(CanalConstants.CANAL_SERVER_MODE, "tcp"); + if (serverMode.equalsIgnoreCase("kafka")) { + canalServerStarter = new CanalKafkaStarter(); + } else if (serverMode.equalsIgnoreCase("rocketMQ")) { + // 预留rocketMQ启动 + } + if (canalServerStarter != null) { + canalServerStarter.init(); + } } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); System.exit(0); diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 9083477d..56c35564 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -8,6 +8,8 @@ canal.zkServers= # flush data to zk canal.zookeeper.flush.period = 1000 canal.withoutNetty = false +# tcp, kafka, rocketMQ +canal.serverMode = tcp # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} canal.file.flush.period = 1000 diff --git a/kafka/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml similarity index 85% rename from kafka/src/main/resources/kafka.yml rename to deployer/src/main/resources/kafka.yml index 2729fc27..93d4d557 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,8 +12,8 @@ canalDestinations: topic: example partition: # 一个destination可以对应多个topic -# topics: -# - topic: example -# partition: + #topics: + # - topic: example + # partition: diff --git a/kafka/pom.xml b/kafka/pom.xml deleted file mode 100644 index b2587c02..00000000 --- a/kafka/pom.xml +++ /dev/null @@ -1,145 +0,0 @@ - - - 4.0.0 - - canal - com.alibaba.otter - 1.1.0-SNAPSHOT - ../pom.xml - - com.alibaba.otter - canal.kafka - jar - canal kafka module for otter ${project.version} - - - com.alibaba.otter - canal.deployer - ${project.version} - - - - org.yaml - snakeyaml - 1.17 - - - org.apache.kafka - kafka_2.11 - 1.1.1 - - - org.slf4j - slf4j-log4j12 - - - - - - org.jboss.netty - netty - 3.2.2.Final - - - - - - - - maven-jar-plugin - - - true - - - **/logback.xml - **/canal.properties - **/spring/** - **/example/** - **/kafka.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 - ${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/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml deleted file mode 100644 index 2c3e3fff..00000000 --- a/kafka/src/main/assembly/dev.xml +++ /dev/null @@ -1,64 +0,0 @@ - - dist - - dir - - false - - - . - / - - README* - - - - ./src/main/bin - bin - - **/* - - 0755 - - - ../deployer/src/main/conf - /conf - - **/* - - - - ../deployer/src/main/resources - /conf - - **/* - - - logback.xml - - - - ./src/main/resources - /conf - - **/* - - - - target - logs - - **/* - - - - - - lib - - junit:junit - - - - diff --git a/kafka/src/main/assembly/release.xml b/kafka/src/main/assembly/release.xml deleted file mode 100644 index 0b103620..00000000 --- a/kafka/src/main/assembly/release.xml +++ /dev/null @@ -1,64 +0,0 @@ - - dist - - tar.gz - - false - - - . - / - - README* - - - - ./src/main/bin - bin - - **/* - - 0755 - - - ../deployer/src/main/conf - /conf - - **/* - - - - ../deployer/src/main/resources - /conf - - **/* - - - logback.xml - - - - ./src/main/resources - /conf - - **/* - - - - target - logs - - **/* - - - - - - lib - - junit:junit - - - - diff --git a/kafka/src/main/bin/startup.bat b/kafka/src/main/bin/startup.bat deleted file mode 100755 index 429cc653..00000000 --- a/kafka/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.properties -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.kafka.CanalLauncher -echo start cmd : %CMD_STR% - -java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.kafka.CanalLauncher \ No newline at end of file diff --git a/kafka/src/main/bin/startup.sh b/kafka/src/main/bin/startup.sh deleted file mode 100644 index 184c4b3f..00000000 --- a/kafka/src/main/bin/startup.sh +++ /dev/null @@ -1,104 +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.properties -logback_configurationFile=$base/conf/logback.xml -export LANG=en_US.UTF-8 -export BASE=$base - -if [ -f $base/bin/canal.pid ] ; then - echo "found canal.pid , Please run stop.sh first ,then startup.sh" 2>&2 - exit 1 -fi - -if [ ! -d $base/logs/canal ] ; then - mkdir -p $base/logs/canal -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.kafka.CanalLauncher 1>>$base/logs/canal/canal.log 2>&1 & - echo $! > $base/bin/canal.pid - - echo "cd to $current_path for continue" - cd $current_path -else - echo "canal conf("$canal_conf") OR log configration file($logback_configurationFile) is not exist,please create then first!" -fi diff --git a/kafka/src/main/bin/stop.sh b/kafka/src/main/bin/stop.sh deleted file mode 100644 index f398749c..00000000 --- a/kafka/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.pid -if [ ! -f "$pidfile" ];then - echo "canal 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/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java deleted file mode 100644 index 3212733c..00000000 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java +++ /dev/null @@ -1,17 +0,0 @@ -package com.alibaba.otter.canal.kafka; - -import com.alibaba.otter.canal.kafka.producer.CanalKafkaStarter; - -/** - * canal-kafka独立版本启动的入口类 - * - * @author machengyuan 2018-6-11 下午05:30:49 - * @version 1.0.0 - */ -public class CanalLauncher { - - public static void main(String[] args) { - CanalServerStarter.init(); - CanalKafkaStarter.init(); - } -} diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java deleted file mode 100644 index 573f0eac..00000000 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java +++ /dev/null @@ -1,78 +0,0 @@ -package com.alibaba.otter.canal.kafka; - -import java.io.FileInputStream; -import java.util.Properties; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.otter.canal.deployer.CanalController; - -/** - * canal server 启动类 - * - * @author machengyuan 2018-6-11 下午05:30:49 - * @version 1.0.0 - */ -public class CanalServerStarter { - - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalServerStarter.class); - private volatile static boolean running = false; - - public static void init() { - try { - logger.info("## set default uncaught exception handler"); - setGlobalUncaughtExceptionHandler(); - - logger.info("## load canal configurations"); - String conf = System.getProperty("canal.conf", "classpath:canal.properties"); - Properties properties = new Properties(); - if (conf.startsWith(CLASSPATH_URL_PREFIX)) { - conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); - properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); - } else { - properties.load(new FileInputStream(conf)); - } - - logger.info("## start the canal server."); - final CanalController controller = new CanalController(properties); - controller.start(); - running = true; - logger.info("## the canal server is running now ......"); - Runtime.getRuntime().addShutdownHook(new Thread() { - - public void run() { - try { - logger.info("## stop the canal server"); - running = false; - controller.stop(); - } catch (Throwable e) { - logger.warn("##something goes wrong when stopping canal Server:", e); - } finally { - logger.info("## canal server is down."); - } - } - - }); - } catch (Throwable e) { - logger.error("## Something goes wrong when starting up the canal Server:", e); - System.exit(0); - } - } - - public static boolean isRunning() { - return running; - } - - private static void setGlobalUncaughtExceptionHandler() { - Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { - - @Override - public void uncaughtException(Thread t, Throwable e) { - logger.error("UnCaughtException", e); - } - }); - } -} diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml deleted file mode 100644 index 34e7b1e5..00000000 --- a/kafka/src/main/resources/logback.xml +++ /dev/null @@ -1,85 +0,0 @@ - - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n - - - - - - - destination - canal - - - - ../logs/${destination}/${destination}.log - - - ../logs/${destination}/%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 - - - - - - - - - destination - canal - - - - ../logs/${destination}/meta.log - - - ../logs/${destination}/%d{yyyy-MM-dd}/meta-%d{yyyy-MM-dd}-%i.log.gz - - - 32MB - - 60 - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} - %msg%n - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/pom.xml b/pom.xml index e7bb158e..64588a28 100644 --- a/pom.xml +++ b/pom.xml @@ -117,7 +117,6 @@ client deployer example - kafka kafka-client prometheus client-adapter diff --git a/server/pom.xml b/server/pom.xml index b7f83d56..3b8e3ddb 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -25,6 +25,29 @@ canal.instance.manager ${project.version} + + + org.yaml + snakeyaml + 1.17 + + + org.apache.kafka + kafka_2.11 + 1.1.1 + + + org.slf4j + slf4j-log4j12 + + + + + + org.jboss.netty + netty + 3.2.2.Final + diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java similarity index 91% rename from kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java rename to server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 0bb15308..b6dba051 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.kafka.producer; +package com.alibaba.otter.canal.kafka; import java.io.IOException; import java.util.Properties; @@ -10,7 +10,6 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; import com.alibaba.otter.canal.protocol.Message; /** @@ -21,7 +20,7 @@ import com.alibaba.otter.canal.protocol.Message; */ public class CanalKafkaProducer { - private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); + private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); private Producer producer; @@ -49,7 +48,7 @@ public class CanalKafkaProducer { } } - public void send(Topic topic, Message message) throws IOException { + public void send(KafkaProperties.Topic topic, Message message) throws IOException { // set canal.instance.filter.transaction.entry = true // boolean valid = false; diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java similarity index 83% rename from kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java rename to server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index 0eedfa7a..d244ce31 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.kafka.producer; +package com.alibaba.otter.canal.kafka; import java.io.FileInputStream; import java.util.List; @@ -10,11 +10,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; -import com.alibaba.otter.canal.kafka.CanalServerStarter; -import com.alibaba.otter.canal.kafka.producer.KafkaProperties.CanalDestination; -import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; +import com.alibaba.otter.canal.kafka.KafkaProperties.CanalDestination; +import com.alibaba.otter.canal.kafka.KafkaProperties.Topic; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.server.CanalServerStarter; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; /** @@ -23,20 +23,21 @@ import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; * @author machengyuan 2018-6-11 下午05:30:49 * @version 1.0.0 */ -public class CanalKafkaStarter { +public class CanalKafkaStarter implements CanalServerStarter { - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalKafkaStarter.class); + private static final Logger logger = LoggerFactory.getLogger(CanalKafkaStarter.class); - private volatile static boolean running = false; + private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static ExecutorService executorService; + private volatile boolean running = false; - private static CanalKafkaProducer canalKafkaProducer; + private ExecutorService executorService; - private static KafkaProperties kafkaProperties; + private CanalKafkaProducer canalKafkaProducer; - public static void init() { + private KafkaProperties kafkaProperties; + + public void init() { try { logger.info("## load kafka configurations"); String conf = System.getProperty("kafka.conf", "classpath:kafka.yml"); @@ -96,11 +97,9 @@ public class CanalKafkaStarter { } } - private static void worker(CanalDestination destination) { + private void worker(CanalDestination destination) { while (!running) ; - while (!CanalServerStarter.isRunning()) - ; // 等待server启动完成 logger.info("## start the canal consumer: {}.", destination.getCanalDestination()); CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); @@ -121,7 +120,7 @@ public class CanalKafkaStarter { Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据 long batchId = message.getId(); try { - int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); + int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { if (!StringUtils.isEmpty(destination.getTopic())) { Topic topic = new Topic(); diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java similarity index 98% rename from kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java rename to server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java index 8e193aac..140ce965 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.kafka.producer; +package com.alibaba.otter.canal.kafka; import java.util.ArrayList; import java.util.HashSet; diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java similarity index 83% rename from kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java rename to server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java index 02336004..49db5aa1 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.kafka.producer; +package com.alibaba.otter.canal.kafka; import java.util.List; import java.util.Map; @@ -37,20 +37,20 @@ public class MessageSerializer implements Serializer { List rowEntries = data.getRawEntries(); // message size int messageSize = 0; - messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, data.getId()); + messageSize += CodedOutputStream.computeInt64Size(1, data.getId()); int dataSize = 0; for (int i = 0; i < rowEntries.size(); i++) { - dataSize += com.google.protobuf.CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); + dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); } messageSize += dataSize; messageSize += 1 * rowEntries.size(); // packet size int size = 0; - size += com.google.protobuf.CodedOutputStream.computeEnumSize(3, + size += CodedOutputStream.computeEnumSize(3, PacketType.MESSAGES.getNumber()); - size += com.google.protobuf.CodedOutputStream.computeTagSize(5) - + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize) + size += CodedOutputStream.computeTagSize(5) + + CodedOutputStream.computeRawVarint32Size(messageSize) + messageSize; // build data byte[] body = new byte[size]; @@ -73,7 +73,7 @@ public class MessageSerializer implements Serializer { } CanalPacket.Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder(); - packetBuilder.setType(CanalPacket.PacketType.MESSAGES); + packetBuilder.setType(PacketType.MESSAGES); packetBuilder.setBody(messageBuilder.build().toByteString()); return packetBuilder.build().toByteArray(); } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalServerStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalServerStarter.java new file mode 100644 index 00000000..18f3f7d4 --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalServerStarter.java @@ -0,0 +1,12 @@ +package com.alibaba.otter.canal.server; + +/** + * 外部服务如Kafka, RocketMQ启动接口 + * + * @author machengyuan 2018-8-23 下午05:20:29 + * @version 1.0.0 + */ +public interface CanalServerStarter { + + void init(); +}