From 31f72e1aac99a9f0e4f0bbeb55c2875625c6bbb7 Mon Sep 17 00:00:00 2001 From: rewerma Date: Mon, 11 Jun 2018 19:52:55 +0800 Subject: [PATCH 01/27] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=20kafaka=20producer=20?= =?UTF-8?q?=E7=9B=B4=E6=8E=A5=E4=BB=8Ecanal=20server=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/deployer/CanalConstants.java | 1 + .../otter/canal/deployer/CanalController.java | 21 ++- deployer/src/main/resources/canal.properties | 1 + kafka/pom.xml | 131 ++++++++++++++++++ kafka/src/main/assembly/dev.xml | 61 ++++++++ kafka/src/main/assembly/release.xml | 61 ++++++++ kafka/src/main/bin/startup.bat | 25 ++++ kafka/src/main/bin/startup.sh | 104 ++++++++++++++ kafka/src/main/bin/stop.sh | 65 +++++++++ .../otter/canal/kafka/CanalLauncher.java | 70 ++++++++++ .../kafka/producer/CanalKafkaProducer.java | 62 +++++++++ .../kafka/producer/CanalKafkaStarter.java | 88 ++++++++++++ .../canal/kafka/producer/KafkaProperties.java | 111 +++++++++++++++ .../kafka/producer/MessageSerializer.java | 49 +++++++ kafka/src/main/resources/kafka.yml | 12 ++ pom.xml | 3 +- 16 files changed, 857 insertions(+), 8 deletions(-) create mode 100644 kafka/pom.xml create mode 100644 kafka/src/main/assembly/dev.xml create mode 100644 kafka/src/main/assembly/release.xml create mode 100755 kafka/src/main/bin/startup.bat create mode 100644 kafka/src/main/bin/startup.sh create mode 100644 kafka/src/main/bin/stop.sh create mode 100644 kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java create mode 100644 kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java create mode 100644 kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java create mode 100644 kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java create mode 100644 kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java create mode 100644 kafka/src/main/resources/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 d68d5871..b328feb7 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 @@ -16,6 +16,7 @@ public class CanalConstants { public static final String CANAL_IP = ROOT + "." + "ip"; public static final String CANAL_PORT = ROOT + "." + "port"; public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers"; + public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty"; public static final String CANAL_DESTINATIONS = ROOT + "." + "destinations"; public static final String CANAL_AUTO_SCAN = ROOT + "." + "auto.scan"; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index 08c0744d..7bd78c5f 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -43,7 +43,7 @@ import com.google.common.collect.MigrateMap; /** * canal调度控制器 - * + * * @author jianghang 2012-11-8 下午12:03:11 * @version 1.0.0 */ @@ -97,9 +97,12 @@ public class CanalController { port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT)); embededCanalServer = CanalServerWithEmbedded.instance(); embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator - canalServer = CanalServerWithNetty.instance(); - canalServer.setIp(ip); - canalServer.setPort(port); + String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY); + if (canalWithoutNetty == null || "true".equals(canalWithoutNetty)) { + canalServer = CanalServerWithNetty.instance(); + canalServer.setIp(ip); + canalServer.setPort(port); + } // 处理下ip为空，默认使用hostIp暴露到zk中 if (StringUtils.isEmpty(ip)) { @@ -431,11 +434,15 @@ public class CanalController { } // 启动网络接口 - canalServer.start(); + if(canalServer!=null) { + canalServer.start(); + } } public void stop() throws Throwable { - canalServer.stop(); + if(canalServer!=null) { + canalServer.stop(); + } if (autoScan) { for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) { @@ -454,7 +461,7 @@ public class CanalController { // 释放canal的工作节点 releaseCid(ZookeeperPathUtils.getCanalClusterNode(ip + ":" + port)); logger.info("## stop the canal server[{}:{}]", ip, port); - + if (zkclientx != null) { zkclientx.close(); } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index f765e71b..c2f57375 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -7,6 +7,7 @@ canal.port= 11111 canal.zkServers= # flush data to zk canal.zookeeper.flush.period = 1000 +canal.withoutNetty = false # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} canal.file.flush.period = 1000 diff --git a/kafka/pom.xml b/kafka/pom.xml new file mode 100644 index 00000000..67397a19 --- /dev/null +++ b/kafka/pom.xml @@ -0,0 +1,131 @@ + + + + canal + com.alibaba.otter + 1.0.26-SNAPSHOT + + 4.0.0 + + canal.kafka + + + + com.alibaba.otter + canal.deployer + ${project.version} + + + + + org.apache.kafka + kafka-clients + 0.10.0.1 + + + org.apache.kafka + kafka_2.11 + 0.10.0.1 + + + + + + + + maven-jar-plugin + + + true + + + **/logback.xml + **/canal.properties + **/spring/** + **/example/** + + + + + + 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} + + + + + + + \ No newline at end of file diff --git a/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml new file mode 100644 index 00000000..df9eca81 --- /dev/null +++ b/kafka/src/main/assembly/dev.xml @@ -0,0 +1,61 @@ + + dist + + dir + + false + + + . + / + + README* + + + + ./src/main/bin + bin + + **/* + + 0755 + + + ../deployer/src/main/conf + /conf + + **/* + + + + ../deployer/src/main/resources + /conf + + **/* + + + + ./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 new file mode 100644 index 00000000..d79ecf3c --- /dev/null +++ b/kafka/src/main/assembly/release.xml @@ -0,0 +1,61 @@ + + dist + + tar.gz + + false + + + . + / + + README* + + + + ./src/main/bin + bin + + **/* + + 0755 + + + ../deployer/src/main/conf + /conf + + **/* + + + + ../deployer/src/main/resources + /conf + + **/* + + + + ./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 new file mode 100755 index 00000000..429cc653 --- /dev/null +++ b/kafka/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.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 new file mode 100644 index 00000000..184c4b3f --- /dev/null +++ b/kafka/src/main/bin/startup.sh @@ -0,0 +1,104 @@ +#!/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 new file mode 100644 index 00000000..f398749c --- /dev/null +++ b/kafka/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.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 new file mode 100644 index 00000000..0797004e --- /dev/null +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java @@ -0,0 +1,70 @@ +package com.alibaba.otter.canal.kafka; + + +import com.alibaba.otter.canal.deployer.CanalController; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileInputStream; +import java.util.Properties; + +/** + * canal-kafka独立版本启动的入口类 + * + * @author machengyuan 2018-6-11 下午05:30:49 + * @version 1.0.0 + */ +public class CanalLauncher { + + private static final String CLASSPATH_URL_PREFIX = "classpath:"; + private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + + public static void main(String[] args) throws Throwable { + 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(com.alibaba.otter.canal.deployer.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(); + logger.info("## the canal server is running now ......"); + Runtime.getRuntime().addShutdownHook(new Thread() { + + public void run() { + try { + logger.info("## stop the canal server"); + 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); + } + } + + private static void setGlobalUncaughtExceptionHandler() { + Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread t, Throwable e) { + logger.error("UnCaughtException", e); + } + }); + } +} \ No newline at end of file diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java new file mode 100644 index 00000000..9a64b1a5 --- /dev/null +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -0,0 +1,62 @@ +package com.alibaba.otter.canal.kafka.producer; + +import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; +import com.alibaba.otter.canal.protocol.Message; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Properties; + +/** + * kafka producer 主操作类 + * + * @author machengyuan 2018-6-11 下午05:30:49 + * @version 1.0.0 + */ +public class CanalKafkaProducer { + private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); + + private static Producer producer; + + public static void init(KafkaProperties kafkaProperties) { + Properties properties = new Properties(); + properties.put("bootstrap.servers", kafkaProperties.getServers()); + properties.put("acks", "all"); + properties.put("retries", kafkaProperties.getRetries()); + properties.put("batch.size", kafkaProperties.getBatchSize()); + properties.put("linger.ms", kafkaProperties.getLingerMs()); + properties.put("buffer.memory", kafkaProperties.getBufferMemory()); + properties.put("key.serializer", StringSerializer.class.getName()); + properties.put("value.serializer", MessageSerializer.class.getName()); + producer = new KafkaProducer(properties); + + Runtime.getRuntime().addShutdownHook(new Thread() { + + public void run() { + try { + logger.info("## stop the kafka producer"); + producer.close(); + } catch (Throwable e) { + logger.warn("##something goes wrong when stopping kafka producer:", e); + } finally { + logger.info("## kafka producer is down."); + } + } + + }); + } + + public static void send(Topic topic, Message message) { + ProducerRecord record; + if (topic.getPartition() != null) { + record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + } else { + record = new ProducerRecord(topic.getTopic(), message); + } + producer.send(record); + } +} diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java new file mode 100644 index 00000000..bc048956 --- /dev/null +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -0,0 +1,88 @@ +package com.alibaba.otter.canal.kafka.producer; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; +import com.alibaba.otter.canal.protocol.ClientIdentity; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class CanalKafkaStarter { + private static final Logger logger = LoggerFactory.getLogger(CanalKafkaStarter.class); + + private volatile static boolean running = false; + private static short CLIENT_ID = 1001; + + public static void init(KafkaProperties kafkaProperties) { + //初始化 kafka producer + CanalKafkaProducer.init(kafkaProperties); + + //对应每个instance启动一个worker线程 + List topics = kafkaProperties.getTopics(); + + ExecutorService executorService = Executors.newFixedThreadPool(topics.size()); + + for (final Topic topic : topics) { + executorService.execute(new Runnable() { + @Override + public void run() { + worker(topic); + } + }); + } + running = true; + } + + private static void sleep(long time) { + try { + Thread.sleep(time); + } catch (InterruptedException e) { + //ignore + } + } + + public static void worker(Topic topic) { + while (!running) ; + //TODO 等待canal server启动完毕 + logger.info("## start the canal consumer: {}.", topic.getDestination()); + CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); + ClientIdentity clientIdentity = new ClientIdentity(topic.getDestination(), CLIENT_ID, ""); + while (running) { + try { + if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { + sleep(3000); + continue; + } + server.subscribe(clientIdentity); + logger.info("## the canal consumer {} is running now ......", topic.getDestination()); + + while (running) { + Message message = server.getWithoutAck(clientIdentity, 5 * 1024); // 获取指定数量的数据 + long batchId = message.getId(); + try { + int size = message.getEntries().size(); + if (batchId == -1 || size == 0) { + sleep(1000); + } else { + CanalKafkaProducer.send(topic, message); + } + + if (batchId != -1) { + server.ack(clientIdentity, batchId); // 提交确认 + } + } catch (Exception e) { + server.rollback(clientIdentity); + logger.error(e.getMessage(), e); + sleep(1000); + } + } + } catch (Exception e) { + logger.error("process error!", e); + } + } + } +} diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java new file mode 100644 index 00000000..4b81238e --- /dev/null +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java @@ -0,0 +1,111 @@ +package com.alibaba.otter.canal.kafka.producer; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * kafka 配置项 + * + * @author machengyuan 2018-6-11 下午05:30:49 + * @version 1.0.0 + */ +public class KafkaProperties { + private String servers = "localhost:6667"; + private int retries = 0; + private int batchSize = 16384; + private int lingerMs = 1; + private long bufferMemory = 33554432L; + + private List topics = new ArrayList(); + private Map topicMap = new HashMap(); + + public static class Topic { + private String topic; + private Integer partition; + private String destination; + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public Integer getPartition() { + return partition; + } + + public void setPartition(Integer partition) { + this.partition = partition; + } + + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + } + + public String getServers() { + return servers; + } + + public void setServers(String servers) { + this.servers = servers; + } + + public int getRetries() { + return retries; + } + + public void setRetries(int retries) { + this.retries = retries; + } + + public int getBatchSize() { + return batchSize; + } + + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } + + public int getLingerMs() { + return lingerMs; + } + + public void setLingerMs(int lingerMs) { + this.lingerMs = lingerMs; + } + + public long getBufferMemory() { + return bufferMemory; + } + + public void setBufferMemory(long bufferMemory) { + this.bufferMemory = bufferMemory; + } + + public List getTopics() { + return topics; + } + + public void setTopics(List topics) { + this.topics = topics; + + if (topics != null) { + for (Topic topic : topics) { + this.topicMap.put(topic.destination, topic); + } + } + } + + public Topic getTopicByDestination(String destination) { + return this.topicMap.get(destination); + } +} diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java new file mode 100644 index 00000000..d1a9577a --- /dev/null +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java @@ -0,0 +1,49 @@ +package com.alibaba.otter.canal.kafka.producer; + +import com.alibaba.otter.canal.protocol.CanalEntry; +import com.alibaba.otter.canal.protocol.CanalPacket; +import com.alibaba.otter.canal.protocol.Message; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.serialization.Serializer; +import org.springframework.util.CollectionUtils; + +import java.util.Map; + +/** + * Message类的序列化Kafka实现 + * + * @author machengyuan 2018-6-11 下午05:30:49 + * @version 1.0.0 + */ +public class MessageSerializer implements Serializer { + @Override + public void configure(Map configs, boolean isKey) { + } + + @Override + public byte[] serialize(String topic, Message data) { + try { + if (data == null) + return null; + else { + CanalPacket.Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder(); + if (data.getId() != -1 && !CollectionUtils.isEmpty(data.getEntries())) { + for (CanalEntry.Entry entry : data.getEntries()) { + messageBuilder.addMessages(entry.toByteString()); + } + } + CanalPacket.Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder(); + packetBuilder.setType(CanalPacket.PacketType.MESSAGES); + packetBuilder.setBody(messageBuilder.build().toByteString()); + return packetBuilder.build().toByteArray(); + } + } catch (Exception e) { + throw new SerializationException("Error when serializing message to byte[] "); + } + } + + @Override + public void close() { + // nothing to do + } +} diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml new file mode 100644 index 00000000..2d07cf14 --- /dev/null +++ b/kafka/src/main/resources/kafka.yml @@ -0,0 +1,12 @@ +servers: localhost:6667 +retries: 0 +batchSize: 16384 +lingerMs: 1 +bufferMemory: 33554432 + +topics: + - topic: exampe + partition: + destination: example + + diff --git a/pom.xml b/pom.xml index 1c4ff11b..bbe33de0 100644 --- a/pom.xml +++ b/pom.xml @@ -127,6 +127,7 @@ client deployer example + kafka @@ -254,7 +255,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_371 + 2.0.0_preview_186 com.alibaba From b306bde3ecf31f06675b40e4b01f65bdc2397d22 Mon Sep 17 00:00:00 2001 From: machengyuan4081 Date: Mon, 11 Jun 2018 23:19:28 +0800 Subject: [PATCH 02/27] kafka producer --- kafka/pom.xml | 12 ++ kafka/src/main/assembly/dev.xml | 3 + kafka/src/main/assembly/release.xml | 3 + .../otter/canal/kafka/CanalLauncher.java | 59 +--------- .../otter/canal/kafka/CanalServerStarter.java | 76 ++++++++++++ .../kafka/producer/CanalKafkaProducer.java | 57 ++++++--- .../kafka/producer/CanalKafkaStarter.java | 109 +++++++++++++----- kafka/src/main/resources/kafka.yml | 6 +- kafka/src/main/resources/logback.xml | 85 ++++++++++++++ 9 files changed, 305 insertions(+), 105 deletions(-) create mode 100644 kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java create mode 100644 kafka/src/main/resources/logback.xml diff --git a/kafka/pom.xml b/kafka/pom.xml index 67397a19..5c015e56 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -18,6 +18,12 @@ ${project.version} + + org.yaml + snakeyaml + 1.17 + + org.apache.kafka @@ -28,6 +34,12 @@ org.apache.kafka kafka_2.11 0.10.0.1 + + + org.slf4j + slf4j-log4j12 + + diff --git a/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml index df9eca81..2c3e3fff 100644 --- a/kafka/src/main/assembly/dev.xml +++ b/kafka/src/main/assembly/dev.xml @@ -34,6 +34,9 @@ **/* + + logback.xml + ./src/main/resources diff --git a/kafka/src/main/assembly/release.xml b/kafka/src/main/assembly/release.xml index d79ecf3c..0b103620 100644 --- a/kafka/src/main/assembly/release.xml +++ b/kafka/src/main/assembly/release.xml @@ -34,6 +34,9 @@ **/* + + logback.xml + ./src/main/resources 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 index 0797004e..c38dc968 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java @@ -1,13 +1,7 @@ package com.alibaba.otter.canal.kafka; -import com.alibaba.otter.canal.deployer.CanalController; -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.FileInputStream; -import java.util.Properties; +import com.alibaba.otter.canal.kafka.producer.CanalKafkaStarter; /** * canal-kafka独立版本启动的入口类 @@ -17,54 +11,9 @@ import java.util.Properties; */ public class CanalLauncher { - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + public static void main(String[] args) { + CanalServerStarter.init(); - public static void main(String[] args) throws Throwable { - 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(com.alibaba.otter.canal.deployer.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(); - logger.info("## the canal server is running now ......"); - Runtime.getRuntime().addShutdownHook(new Thread() { - - public void run() { - try { - logger.info("## stop the canal server"); - 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); - } - } - - private static void setGlobalUncaughtExceptionHandler() { - Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { - @Override - public void uncaughtException(Thread t, Throwable e) { - logger.error("UnCaughtException", e); - } - }); + CanalKafkaStarter.init(); } } \ No newline at end of file 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 new file mode 100644 index 00000000..0b033716 --- /dev/null +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java @@ -0,0 +1,76 @@ +package com.alibaba.otter.canal.kafka; + +import com.alibaba.otter.canal.deployer.CanalController; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileInputStream; +import java.util.Properties; + +/** + * 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 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(com.alibaba.otter.canal.deployer.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/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java index 9a64b1a5..ae094b27 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.kafka.producer; import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; +import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; @@ -9,6 +10,7 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.List; import java.util.Properties; /** @@ -22,7 +24,7 @@ public class CanalKafkaProducer { private static Producer producer; - public static void init(KafkaProperties kafkaProperties) { + static void init(KafkaProperties kafkaProperties) { Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", "all"); @@ -33,24 +35,45 @@ public class CanalKafkaProducer { properties.put("key.serializer", StringSerializer.class.getName()); properties.put("value.serializer", MessageSerializer.class.getName()); producer = new KafkaProducer(properties); - - Runtime.getRuntime().addShutdownHook(new Thread() { - - public void run() { - try { - logger.info("## stop the kafka producer"); - producer.close(); - } catch (Throwable e) { - logger.warn("##something goes wrong when stopping kafka producer:", e); - } finally { - logger.info("## kafka producer is down."); - } - } - - }); } - public static void send(Topic topic, Message message) { + static void stop() { + try { + logger.info("## stop the kafka producer"); + producer.close(); + } catch (Throwable e) { + logger.warn("##something goes wrong when stopping kafka producer:", e); + } finally { + logger.info("## kafka producer is down."); + } + } + + static void send(Topic topic, Message message) { + try { + List entries = message.getEntries(); + boolean flag = false; + if (!entries.isEmpty()) { + for (CanalEntry.Entry entry : entries) { + CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + if (rowChage.getIsDdl()) { + flag = true; + break; + } else { + if (!rowChage.getRowDatasList().isEmpty()) { + flag = true; + break; + } + } + } + } + if (!flag) { + return; + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + return; + } + ProducerRecord record; if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index bc048956..a0dc3691 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -1,60 +1,106 @@ package com.alibaba.otter.canal.kafka.producer; +import java.io.FileInputStream; import java.util.List; +import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import com.alibaba.otter.canal.kafka.CanalServerStarter; import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; +import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.yaml.snakeyaml.Yaml; +/** + * kafka 启动类 + * + * @author machengyuan 2018-6-11 下午05:30:49 + * @version 1.0.0 + */ public class CanalKafkaStarter { + private static final String CLASSPATH_URL_PREFIX = "classpath:"; private static final Logger logger = LoggerFactory.getLogger(CanalKafkaStarter.class); private volatile static boolean running = false; - private static short CLIENT_ID = 1001; - public static void init(KafkaProperties kafkaProperties) { - //初始化 kafka producer - CanalKafkaProducer.init(kafkaProperties); + private static ExecutorService executorService; - //对应每个instance启动一个worker线程 - List topics = kafkaProperties.getTopics(); - - ExecutorService executorService = Executors.newFixedThreadPool(topics.size()); - - for (final Topic topic : topics) { - executorService.execute(new Runnable() { - @Override - public void run() { - worker(topic); - } - }); - } - running = true; - } - - private static void sleep(long time) { + public static void init() { try { - Thread.sleep(time); - } catch (InterruptedException e) { - //ignore + + logger.info("## load kafka configurations"); + String conf = System.getProperty("canal.conf", "classpath:kafka.yml"); + + KafkaProperties kafkaProperties; + if (conf.startsWith(CLASSPATH_URL_PREFIX)) { + conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); + kafkaProperties = new Yaml().loadAs(CanalKafkaStarter.class.getClassLoader().getResourceAsStream(conf), KafkaProperties.class); + } else { + kafkaProperties = new Yaml().loadAs(new FileInputStream(conf), KafkaProperties.class); + } + + //初始化 kafka producer + CanalKafkaProducer.init(kafkaProperties); + + //对应每个instance启动一个worker线程 + List topics = kafkaProperties.getTopics(); + + executorService = Executors.newFixedThreadPool(topics.size()); + + logger.info("## start the kafka workers."); + for (final Topic topic : topics) { + executorService.execute(new Runnable() { + @Override + public void run() { + worker(topic); + } + }); + } + running = true; + logger.info("## the kafka workers is running now ......"); + Runtime.getRuntime().addShutdownHook(new Thread() { + public void run() { + try { + logger.info("## stop the kafka workers"); + running = false; + executorService.shutdown(); + CanalKafkaProducer.stop(); + } catch (Throwable e) { + logger.warn("##something goes wrong when stopping kafka workers:", e); + } finally { + logger.info("## canal kafka is down."); + } + } + + }); + + } catch (Throwable e) { + logger.error("## Something goes wrong when starting up the canal kafka workers:", e); + System.exit(0); } } - public static void worker(Topic topic) { + + private static void worker(Topic topic) { while (!running) ; - //TODO 等待canal server启动完毕 + while (!CanalServerStarter.isRunning()) ; //等待server启动完成 logger.info("## start the canal consumer: {}.", topic.getDestination()); CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); - ClientIdentity clientIdentity = new ClientIdentity(topic.getDestination(), CLIENT_ID, ""); + ClientIdentity clientIdentity = new ClientIdentity(topic.getDestination(), (short) 1001, ""); while (running) { try { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { - sleep(3000); + try { + Thread.sleep(3000); + } catch (InterruptedException e) { + //ignore + } continue; } server.subscribe(clientIdentity); @@ -66,7 +112,11 @@ public class CanalKafkaStarter { try { int size = message.getEntries().size(); if (batchId == -1 || size == 0) { - sleep(1000); + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + //ignore + } } else { CanalKafkaProducer.send(topic, message); } @@ -77,7 +127,6 @@ public class CanalKafkaStarter { } catch (Exception e) { server.rollback(clientIdentity); logger.error(e.getMessage(), e); - sleep(1000); } } } catch (Exception e) { diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index 2d07cf14..af84977d 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -1,12 +1,12 @@ -servers: localhost:6667 +servers: slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667 retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 topics: - - topic: exampe + - topic: example2 partition: - destination: example + destination: example2 diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml new file mode 100644 index 00000000..158c1699 --- /dev/null +++ b/kafka/src/main/resources/logback.xml @@ -0,0 +1,85 @@ + + + + + %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 From 9b7d62717ea210d933347a712e38a08c02a04ddd Mon Sep 17 00:00:00 2001 From: machengyuan4081 Date: Mon, 11 Jun 2018 23:31:02 +0800 Subject: [PATCH 03/27] =?UTF-8?q?kafka=20system.properties=20=E7=9A=84=20b?= =?UTF-8?q?ug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/pom.xml | 2 ++ .../alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/kafka/pom.xml b/kafka/pom.xml index 5c015e56..4cb0b9bf 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -57,6 +57,8 @@ **/canal.properties **/spring/** **/example/** + **/example2/** + **/kafka.yml diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index a0dc3691..e89daefc 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -35,7 +35,7 @@ public class CanalKafkaStarter { try { logger.info("## load kafka configurations"); - String conf = System.getProperty("canal.conf", "classpath:kafka.yml"); + String conf = System.getProperty("kafka.conf", "classpath:kafka.yml"); KafkaProperties kafkaProperties; if (conf.startsWith(CLASSPATH_URL_PREFIX)) { From e4cde2f6a12d5d2a44de16baf0274dc2b162c0cd Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 09:00:55 +0800 Subject: [PATCH 04/27] =?UTF-8?q?=E4=BF=AE=E6=94=B9withoutNetty=E7=9A=84bu?= =?UTF-8?q?g?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/com/alibaba/otter/canal/deployer/CanalController.java | 2 +- .../java/com/alibaba/otter/canal/kafka/CanalServerStarter.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index 7bd78c5f..9b866071 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -98,7 +98,7 @@ public class CanalController { embededCanalServer = CanalServerWithEmbedded.instance(); embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY); - if (canalWithoutNetty == null || "true".equals(canalWithoutNetty)) { + if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) { canalServer = CanalServerWithNetty.instance(); canalServer.setIp(ip); canalServer.setPort(port); 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 index 0b033716..549ec17f 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java @@ -30,7 +30,7 @@ public class CanalServerStarter { Properties properties = new Properties(); if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); - properties.load(com.alibaba.otter.canal.deployer.CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); + properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); } else { properties.load(new FileInputStream(conf)); } From 0d6757c2572c1611ff119cc6e0b2a8d806ba5843 Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 09:46:01 +0800 Subject: [PATCH 05/27] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=E7=9A=84=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/logback.xml | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/deployer/src/main/resources/logback.xml b/deployer/src/main/resources/logback.xml index 1b0f4103..f1c0e93b 100644 --- a/deployer/src/main/resources/logback.xml +++ b/deployer/src/main/resources/logback.xml @@ -70,9 +70,13 @@ - + + + + + From e35a06743c69831e71366f7b607abba9dc3cda7f Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 09:46:54 +0800 Subject: [PATCH 06/27] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=E7=9A=84=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/src/main/assembly/dev.xml | 3 - kafka/src/main/assembly/release.xml | 3 - .../kafka/producer/CanalKafkaProducer.java | 31 ++----- .../kafka/producer/CanalKafkaStarter.java | 21 ++--- .../canal/kafka/producer/KafkaProperties.java | 13 --- kafka/src/main/resources/kafka.yml | 4 +- kafka/src/main/resources/logback.xml | 85 ------------------- 7 files changed, 18 insertions(+), 142 deletions(-) delete mode 100644 kafka/src/main/resources/logback.xml diff --git a/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml index 2c3e3fff..df9eca81 100644 --- a/kafka/src/main/assembly/dev.xml +++ b/kafka/src/main/assembly/dev.xml @@ -34,9 +34,6 @@ **/* - - logback.xml - ./src/main/resources diff --git a/kafka/src/main/assembly/release.xml b/kafka/src/main/assembly/release.xml index 0b103620..d79ecf3c 100644 --- a/kafka/src/main/assembly/release.xml +++ b/kafka/src/main/assembly/release.xml @@ -34,9 +34,6 @@ **/* - - logback.xml - ./src/main/resources diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java index ae094b27..af2bcfca 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -22,9 +22,9 @@ import java.util.Properties; public class CanalKafkaProducer { private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); - private static Producer producer; + private Producer producer; - static void init(KafkaProperties kafkaProperties) { + public void init(KafkaProperties kafkaProperties) { Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", "all"); @@ -37,7 +37,7 @@ public class CanalKafkaProducer { producer = new KafkaProducer(properties); } - static void stop() { + public void stop() { try { logger.info("## stop the kafka producer"); producer.close(); @@ -48,29 +48,8 @@ public class CanalKafkaProducer { } } - static void send(Topic topic, Message message) { - try { - List entries = message.getEntries(); - boolean flag = false; - if (!entries.isEmpty()) { - for (CanalEntry.Entry entry : entries) { - CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - if (rowChage.getIsDdl()) { - flag = true; - break; - } else { - if (!rowChage.getRowDatasList().isEmpty()) { - flag = true; - break; - } - } - } - } - if (!flag) { - return; - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + public void send(Topic topic, Message message) { + if (message == null || message.getEntries().isEmpty()) { return; } diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index e89daefc..7514c5ce 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -1,14 +1,7 @@ package com.alibaba.otter.canal.kafka.producer; -import java.io.FileInputStream; -import java.util.List; -import java.util.Properties; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - import com.alibaba.otter.canal.kafka.CanalServerStarter; import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; -import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; @@ -17,6 +10,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; +import java.io.FileInputStream; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + /** * kafka 启动类 * @@ -31,6 +29,8 @@ public class CanalKafkaStarter { private static ExecutorService executorService; + private static CanalKafkaProducer canalKafkaProducer; + public static void init() { try { @@ -46,7 +46,8 @@ public class CanalKafkaStarter { } //初始化 kafka producer - CanalKafkaProducer.init(kafkaProperties); + canalKafkaProducer = new CanalKafkaProducer(); + canalKafkaProducer.init(kafkaProperties); //对应每个instance启动一个worker线程 List topics = kafkaProperties.getTopics(); @@ -70,7 +71,7 @@ public class CanalKafkaStarter { logger.info("## stop the kafka workers"); running = false; executorService.shutdown(); - CanalKafkaProducer.stop(); + canalKafkaProducer.stop(); } catch (Throwable e) { logger.warn("##something goes wrong when stopping kafka workers:", e); } finally { @@ -118,7 +119,7 @@ public class CanalKafkaStarter { //ignore } } else { - CanalKafkaProducer.send(topic, message); + canalKafkaProducer.send(topic, message); } if (batchId != -1) { diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java index 4b81238e..f70ac2c4 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java @@ -1,9 +1,7 @@ package com.alibaba.otter.canal.kafka.producer; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; /** * kafka 配置项 @@ -19,7 +17,6 @@ public class KafkaProperties { private long bufferMemory = 33554432L; private List topics = new ArrayList(); - private Map topicMap = new HashMap(); public static class Topic { private String topic; @@ -97,15 +94,5 @@ public class KafkaProperties { public void setTopics(List topics) { this.topics = topics; - - if (topics != null) { - for (Topic topic : topics) { - this.topicMap.put(topic.destination, topic); - } - } - } - - public Topic getTopicByDestination(String destination) { - return this.topicMap.get(destination); } } diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index af84977d..3a846c61 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -5,8 +5,8 @@ lingerMs: 1 bufferMemory: 33554432 topics: - - topic: example2 + - topic: example partition: - destination: example2 + destination: example diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml deleted file mode 100644 index 158c1699..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 From 330da50a5ecdfd1bb0d9fb1b064e2918d950e392 Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 10:23:50 +0800 Subject: [PATCH 07/27] =?UTF-8?q?=E4=BF=AE=E6=94=B9kafka=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E9=A1=B9=E5=B1=9E=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/producer/CanalKafkaStarter.java | 6 +++--- .../otter/canal/kafka/producer/KafkaProperties.java | 10 +++++----- kafka/src/main/resources/kafka.yml | 2 +- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index 7514c5ce..3ef05489 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -91,9 +91,9 @@ public class CanalKafkaStarter { private static void worker(Topic topic) { while (!running) ; while (!CanalServerStarter.isRunning()) ; //等待server启动完成 - logger.info("## start the canal consumer: {}.", topic.getDestination()); + logger.info("## start the canal consumer: {}.", topic.getCanalDestination()); CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); - ClientIdentity clientIdentity = new ClientIdentity(topic.getDestination(), (short) 1001, ""); + ClientIdentity clientIdentity = new ClientIdentity(topic.getCanalDestination(), (short) 1001, ""); while (running) { try { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { @@ -105,7 +105,7 @@ public class CanalKafkaStarter { continue; } server.subscribe(clientIdentity); - logger.info("## the canal consumer {} is running now ......", topic.getDestination()); + logger.info("## the canal consumer {} is running now ......", topic.getCanalDestination()); while (running) { Message message = server.getWithoutAck(clientIdentity, 5 * 1024); // 获取指定数量的数据 diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java index f70ac2c4..f7362a78 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java @@ -21,7 +21,7 @@ public class KafkaProperties { public static class Topic { private String topic; private Integer partition; - private String destination; + private String canalDestination; public String getTopic() { return topic; @@ -39,12 +39,12 @@ public class KafkaProperties { this.partition = partition; } - public String getDestination() { - return destination; + public String getCanalDestination() { + return canalDestination; } - public void setDestination(String destination) { - this.destination = destination; + public void setCanalDestination(String canalDestination) { + this.canalDestination = canalDestination; } } diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index 3a846c61..7ca1bb7f 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -7,6 +7,6 @@ bufferMemory: 33554432 topics: - topic: example partition: - destination: example + canalDestination: example From 3e97e6b77115ef5713cbaee7f470a900d0334a93 Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 10:53:46 +0800 Subject: [PATCH 08/27] =?UTF-8?q?=E6=95=B4=E7=90=86pom=E6=A0=BC=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/pom.xml | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/kafka/pom.xml b/kafka/pom.xml index 4cb0b9bf..0ca697ba 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -2,15 +2,17 @@ + 4.0.0 canal com.alibaba.otter 1.0.26-SNAPSHOT + ../pom.xml - 4.0.0 - + com.alibaba.otter canal.kafka - + jar + canal kafka module for otter ${project.version} com.alibaba.otter From 6faa3fd586eb902bc78d83c0e55ea17b99f15503 Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 13:28:49 +0800 Subject: [PATCH 09/27] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=AF=B9message?= =?UTF-8?q?=E7=9A=84=E9=AA=8C=E8=AF=81=EF=BC=8C=E8=BF=87=E6=BB=A4transacti?= =?UTF-8?q?on=20begin=20end=20=E7=B1=BB=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/pom.xml | 10 +++++----- .../canal/kafka/producer/CanalKafkaProducer.java | 13 +++++++++++-- 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/kafka/pom.xml b/kafka/pom.xml index 0ca697ba..55e72553 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -27,11 +27,11 @@ - - org.apache.kafka - kafka-clients - 0.10.0.1 - + + + + + org.apache.kafka kafka_2.11 diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java index af2bcfca..21527291 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -10,7 +10,6 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; import java.util.Properties; /** @@ -49,10 +48,20 @@ public class CanalKafkaProducer { } public void send(Topic topic, Message message) { - if (message == null || message.getEntries().isEmpty()) { + boolean valid = false; + if (message != null && !message.getEntries().isEmpty()) { + for (CanalEntry.Entry entry : message.getEntries()) { + if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) { + valid = true; + break; + } + } + } + if (!valid) { return; } + ProducerRecord record; if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); From e7846085fc60f382ce6f6bf899e646c3d03ffa8f Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 19:25:03 +0800 Subject: [PATCH 10/27] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E6=B6=88=E8=B4=B9=E5=AE=8C=E6=88=90=E5=9F=BA=E6=9C=AC=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka-client/pom.xml | 110 +++++++++++ .../kafka/client/KafkaCanalConnector.java | 172 ++++++++++++++++++ .../kafka/client/KafkaCanalConnectors.java | 30 +++ .../kafka/client/MessageDeserializer.java | 55 ++++++ .../client/running/AbstractKafkaTest.java | 19 ++ .../client/running/ClientRunningTest.java | 42 +++++ kafka-client/src/test/resources/logback.xml | 19 ++ kafka/pom.xml | 6 - kafka/src/main/resources/kafka.yml | 5 +- pom.xml | 1 + 10 files changed, 452 insertions(+), 7 deletions(-) create mode 100644 kafka-client/pom.xml create mode 100644 kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java create mode 100644 kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java create mode 100644 kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java create mode 100644 kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java create mode 100644 kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java create mode 100644 kafka-client/src/test/resources/logback.xml diff --git a/kafka-client/pom.xml b/kafka-client/pom.xml new file mode 100644 index 00000000..0efd13d4 --- /dev/null +++ b/kafka-client/pom.xml @@ -0,0 +1,110 @@ + + + 4.0.0 + + canal + com.alibaba.otter + 1.0.26-SNAPSHOT + + com.alibaba.otter + canal.kafka.client + jar + canal kafka client module for otter ${project.version} + + + + + com.alibaba.otter + canal.client + ${project.version} + + + org.apache.kafka + kafka-clients + 0.10.0.1 + + + + + junit + junit + + + + + + dev + + true + + env + !javadoc + + + + + + javadoc + + + env + javadoc + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + 2.9.1 + + + attach-javadocs + package + + jar + + + + + true + public + true +
${project.artifactId}-${project.version}
+
${project.artifactId}-${project.version}
+ ${project.artifactId}-${project.version} + + https://github.com/alibaba/canal + + ${project.build.directory}/apidocs/apidocs/${project.version} + + + + org.apache.maven.plugins + maven-scm-publish-plugin + 1.0-beta-2 + + + attach-javadocs + package + + publish-scm + + + + + ${project.build.directory}/scmpublish + Publishing javadoc for ${project.artifactId}:${project.version} + ${project.build.directory}/apidocs + true + scm:git:git@github.com:alibaba/canal.git + gh-pages + + + + + + + \ No newline at end of file diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java new file mode 100644 index 00000000..fc749fec --- /dev/null +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java @@ -0,0 +1,172 @@ +package com.alibaba.otter.canal.kafka.client; + +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.exception.CanalClientException; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.serialization.StringDeserializer; + +import java.util.Collections; +import java.util.ConcurrentModificationException; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +/** + * canal kafka 数据操作客户端 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ +public class KafkaCanalConnector implements CanalConnector { + + private KafkaConsumer kafkaConsumer; + + private String topic; + + private Integer partition; + + + private Properties properties; + + public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId) { + this.topic = topic; + this.partition = partition; + + properties = new Properties(); + properties.put("bootstrap.servers", servers); + properties.put("group.id", groupId); + properties.put("enable.auto.commit", false); + properties.put("auto.commit.interval.ms", 1000); + properties.put("auto.offset.reset", "latest"); //earliest //如果没有offset则从最后的offset开始读 + properties.put("request.timeout.ms", 600000); + properties.put("offsets.commit.timeout.ms", 300000); + properties.put("session.timeout.ms", 30000); + properties.put("max.poll.records", 1); //一次只取一条message + properties.put("key.deserializer", StringDeserializer.class.getName()); + properties.put("value.deserializer", MessageDeserializer.class.getName()); + } + + @Override + public void connect() throws CanalClientException { + kafkaConsumer = new KafkaConsumer(properties); + } + + @Override + public void disconnect() throws CanalClientException { + if (kafkaConsumer != null) { + try { + kafkaConsumer.close(); + } catch (ConcurrentModificationException e) { + kafkaConsumer.wakeup(); //通过wakeup异常间接关闭consumer + } + } + } + + @Override + public boolean checkValid() throws CanalClientException { + return true; + } + + @Override + public void subscribe(String filter) throws CanalClientException { + try { + if (kafkaConsumer == null) { + throw new CanalClientException("connect the kafka first before subscribe"); + } + if (partition == null) { + kafkaConsumer.subscribe(Collections.singletonList(topic)); + } else { + kafkaConsumer.subscribe(Collections.singletonList(topic)); + TopicPartition topicPartition = new TopicPartition(topic, partition); + kafkaConsumer.assign(Collections.singletonList(topicPartition)); + } + } catch (WakeupException e) { + closeByWakeupException(e); + } catch (Exception e) { + throw new CanalClientException(e); + } + } + + @Override + public void subscribe() throws CanalClientException { + subscribe(null); + } + + @Override + public void unsubscribe() throws CanalClientException { + try { + kafkaConsumer.unsubscribe(); + } catch (WakeupException e) { + closeByWakeupException(e); + } catch (Exception e) { + throw new CanalClientException(e); + } + } + + @Override + public Message get(int batchSize) throws CanalClientException { + return get(batchSize, 100L, TimeUnit.MILLISECONDS); + } + + @Override + public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + Message message = getWithoutAck(batchSize, timeout, unit); + this.ack(1); + return message; + } + + @Override + public Message getWithoutAck(int batchSize) throws CanalClientException { + return getWithoutAck(batchSize, 100L, TimeUnit.MILLISECONDS); + } + + @Override + public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + try { + ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，一次最多只能poll到一条Msg + + if (!records.isEmpty()) { + return records.iterator().next().value(); + } + } catch (WakeupException e) { + closeByWakeupException(e); + } catch (Exception e) { + throw new CanalClientException(e); + } + return null; + } + + @Override + public void ack(long batchId) throws CanalClientException { + try { + kafkaConsumer.commitSync(); + } catch (WakeupException e) { + closeByWakeupException(e); + } catch (Exception e) { + throw new CanalClientException(e); + } + } + + @Override + public void rollback(long batchId) throws CanalClientException { + + } + + @Override + public void rollback() throws CanalClientException { + + } + + @Override + public void stopRunning() throws CanalClientException { + + } + + private void closeByWakeupException(WakeupException e) { + kafkaConsumer.close(); + throw e; + } +} diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java new file mode 100644 index 00000000..f0d4dea6 --- /dev/null +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java @@ -0,0 +1,30 @@ +package com.alibaba.otter.canal.kafka.client; + +import com.alibaba.otter.canal.client.CanalConnector; + +public class KafkaCanalConnectors { + /** + * 创建kafka客户端链接 + * + * @param servers + * @param topic + * @param partition + * @param groupId + * @return + */ + public static CanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) { + return new KafkaCanalConnector(servers, topic, partition, groupId); + } + + /** + * 创建kafka客户端链接 + * + * @param servers + * @param topic + * @param groupId + * @return + */ + public static CanalConnector newKafkaConnector(String servers, String topic, String groupId) { + return new KafkaCanalConnector(servers, topic, null, groupId); + } +} diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java new file mode 100644 index 00000000..c321d076 --- /dev/null +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java @@ -0,0 +1,55 @@ +package com.alibaba.otter.canal.kafka.client; + +import com.alibaba.otter.canal.protocol.CanalEntry; +import com.alibaba.otter.canal.protocol.CanalPacket; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.exception.CanalClientException; +import com.google.protobuf.ByteString; +import org.apache.kafka.common.serialization.Deserializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +public class MessageDeserializer implements Deserializer { + private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class); + + @Override + public void configure(Map configs, boolean isKey) { + } + + @Override + public Message deserialize(String topic, byte[] data) { + try { + if (data == null) + return null; + else { + CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data); + switch (p.getType()) { + case MESSAGES: { + if (!p.getCompression().equals(CanalPacket.Compression.NONE)) { + throw new CanalClientException("compression is not supported in this connector"); + } + + CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody()); + Message result = new Message(messages.getBatchId()); + for (ByteString byteString : messages.getMessagesList()) { + result.addEntry(CanalEntry.Entry.parseFrom(byteString)); + } + return result; + } + default: + break; + } + } + } catch (Exception e) { + logger.error("Error when deserializing byte[] to message "); + } + return null; + } + + @Override + public void close() { + // nothing to do + } +} \ No newline at end of file diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java new file mode 100644 index 00000000..2a9ce466 --- /dev/null +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java @@ -0,0 +1,19 @@ +package com.alibaba.otter.canal.kafka.client.running; + +import org.junit.Assert; + +public class AbstractKafkaTest { + + protected String topic = "example"; + protected Integer partition = null; + protected String groupId = "g1"; + protected String servers = "slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667"; + + public void sleep(long time) { + try { + Thread.sleep(time); + } catch (InterruptedException e) { + Assert.fail(e.getMessage()); + } + } +} diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java new file mode 100644 index 00000000..c968f0f3 --- /dev/null +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java @@ -0,0 +1,42 @@ +package com.alibaba.otter.canal.kafka.client.running; + +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors; +import com.alibaba.otter.canal.protocol.Message; +import org.junit.Test; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +public class ClientRunningTest extends AbstractKafkaTest { + + private boolean running = true; + + @Test + public void testKafkaConsumer() { + final ExecutorService executor = Executors.newFixedThreadPool(1); + + final CanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); + kafkaCanalConnector.connect(); + kafkaCanalConnector.subscribe(); + + executor.submit(new Runnable() { + @Override + public void run() { + while (running) { + Message message = kafkaCanalConnector.getWithoutAck(1); + if (message != null) { + System.out.println(message); + sleep(40000); + } + kafkaCanalConnector.ack(1); + } + } + }); + + sleep(120000); + running = false; + kafkaCanalConnector.disconnect(); + } + +} diff --git a/kafka-client/src/test/resources/logback.xml b/kafka-client/src/test/resources/logback.xml new file mode 100644 index 00000000..81fa0712 --- /dev/null +++ b/kafka-client/src/test/resources/logback.xml @@ -0,0 +1,19 @@ + + + + + + %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n + + + + + + + + + + + + + \ No newline at end of file diff --git a/kafka/pom.xml b/kafka/pom.xml index 55e72553..68d6d72e 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -26,12 +26,6 @@ 1.17 - - - - - - org.apache.kafka kafka_2.11 diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index 7ca1bb7f..a869321c 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -5,8 +5,11 @@ lingerMs: 1 bufferMemory: 33554432 topics: - - topic: example + - topic: expTest partition: canalDestination: example +# - topic: example2 +# partition: +# canalDestination: example diff --git a/pom.xml b/pom.xml index bbe33de0..e8a97b45 100644 --- a/pom.xml +++ b/pom.xml @@ -128,6 +128,7 @@ deployer example kafka + kafka-client From d991e6be6086019285d65e9062d97c081319ca54 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 10:34:28 +0800 Subject: [PATCH 11/27] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E6=B6=88=E8=B4=B9=E4=BB=A3=E7=A0=81=E6=95=B4=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka-client/pom.xml | 3 +- .../kafka/client/KafkaCanalConnector.java | 122 ++++++++---------- .../kafka/client/KafkaCanalConnectors.java | 6 +- .../client/running/ClientRunningTest.java | 9 +- kafka/src/main/resources/kafka.yml | 2 +- 5 files changed, 65 insertions(+), 77 deletions(-) diff --git a/kafka-client/pom.xml b/kafka-client/pom.xml index 0efd13d4..264cfa0c 100644 --- a/kafka-client/pom.xml +++ b/kafka-client/pom.xml @@ -7,6 +7,7 @@ canal com.alibaba.otter 1.0.26-SNAPSHOT + ../pom.xml com.alibaba.otter canal.kafka.client @@ -17,7 +18,7 @@ com.alibaba.otter - canal.client + canal.protocol ${project.version} diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java index fc749fec..e3a68934 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java @@ -1,6 +1,5 @@ package com.alibaba.otter.canal.kafka.client; -import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.exception.CanalClientException; import org.apache.kafka.clients.consumer.ConsumerRecords; @@ -20,7 +19,7 @@ import java.util.concurrent.TimeUnit; * @author machengyuan @ 2018-6-12 * @version 1.0.0 */ -public class KafkaCanalConnector implements CanalConnector { +public class KafkaCanalConnector { private KafkaConsumer kafkaConsumer; @@ -39,23 +38,31 @@ public class KafkaCanalConnector implements CanalConnector { properties.put("bootstrap.servers", servers); properties.put("group.id", groupId); properties.put("enable.auto.commit", false); - properties.put("auto.commit.interval.ms", 1000); + properties.put("auto.commit.interval.ms", "1000"); properties.put("auto.offset.reset", "latest"); //earliest //如果没有offset则从最后的offset开始读 - properties.put("request.timeout.ms", 600000); - properties.put("offsets.commit.timeout.ms", 300000); - properties.put("session.timeout.ms", 30000); - properties.put("max.poll.records", 1); //一次只取一条message + properties.put("request.timeout.ms", "360000"); //必须大于session.timeout.ms的设置 + properties.put("session.timeout.ms", "300000"); //默认为5分钟 + properties.put("max.poll.records", "1"); //所以一次只取一条数据 properties.put("key.deserializer", StringDeserializer.class.getName()); properties.put("value.deserializer", MessageDeserializer.class.getName()); } - @Override - public void connect() throws CanalClientException { - kafkaConsumer = new KafkaConsumer(properties); + /** + * 重新设置sessionTime + * + * @param timeout + * @param unit + */ + public void setSessionTimeout(Long timeout, TimeUnit unit) { + long t = unit.toMillis(timeout); + properties.put("request.timeout.ms", String.valueOf(t + 60000)); + properties.put("session.timeout.ms", String.valueOf(t)); } - @Override - public void disconnect() throws CanalClientException { + /** + * 释放链接 + */ + public void disconnect() { if (kafkaConsumer != null) { try { kafkaConsumer.close(); @@ -65,17 +72,17 @@ public class KafkaCanalConnector implements CanalConnector { } } - @Override - public boolean checkValid() throws CanalClientException { - return true; - } - - @Override - public void subscribe(String filter) throws CanalClientException { + /** + * 订阅topic，不能重复订阅 + */ + public void subscribe() { try { - if (kafkaConsumer == null) { - throw new CanalClientException("connect the kafka first before subscribe"); + if (kafkaConsumer != null) { + throw new CanalClientException("resubscribe error"); } + + kafkaConsumer = new KafkaConsumer(properties); + if (partition == null) { kafkaConsumer.subscribe(Collections.singletonList(topic)); } else { @@ -85,86 +92,69 @@ public class KafkaCanalConnector implements CanalConnector { } } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } } - @Override - public void subscribe() throws CanalClientException { - subscribe(null); - } - - @Override - public void unsubscribe() throws CanalClientException { + /** + * 取消订阅 + */ + public void unsubscribe() { try { kafkaConsumer.unsubscribe(); } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } } - @Override - public Message get(int batchSize) throws CanalClientException { - return get(batchSize, 100L, TimeUnit.MILLISECONDS); + /** + * 获取数据，自动进行确认 + * + * @return + */ + public Message get() { + return get(100L, TimeUnit.MILLISECONDS); } - @Override - public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { - Message message = getWithoutAck(batchSize, timeout, unit); - this.ack(1); + public Message get(Long timeout, TimeUnit unit) { + Message message = getWithoutAck(timeout, unit); + this.ack(); return message; } - @Override - public Message getWithoutAck(int batchSize) throws CanalClientException { - return getWithoutAck(batchSize, 100L, TimeUnit.MILLISECONDS); + public Message getWithoutAck() { + return getWithoutAck(100L, TimeUnit.MILLISECONDS); } - @Override - public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + /** + * 获取数据，不进行确认，等待处理完成手工确认 + * + * @return + */ + public Message getWithoutAck(Long timeout, TimeUnit unit) { try { - ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，一次最多只能poll到一条Msg + ConsumerRecords records = + kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据 if (!records.isEmpty()) { return records.iterator().next().value(); } } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } return null; } - @Override - public void ack(long batchId) throws CanalClientException { + /** + * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败 + */ + public void ack() { try { kafkaConsumer.commitSync(); } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } } - @Override - public void rollback(long batchId) throws CanalClientException { - - } - - @Override - public void rollback() throws CanalClientException { - - } - - @Override - public void stopRunning() throws CanalClientException { - - } - private void closeByWakeupException(WakeupException e) { kafkaConsumer.close(); throw e; diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java index f0d4dea6..eb5f248c 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.kafka.client; -import com.alibaba.otter.canal.client.CanalConnector; - public class KafkaCanalConnectors { /** * 创建kafka客户端链接 @@ -12,7 +10,7 @@ public class KafkaCanalConnectors { * @param groupId * @return */ - public static CanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) { + public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) { return new KafkaCanalConnector(servers, topic, partition, groupId); } @@ -24,7 +22,7 @@ public class KafkaCanalConnectors { * @param groupId * @return */ - public static CanalConnector newKafkaConnector(String servers, String topic, String groupId) { + public static KafkaCanalConnector newKafkaConnector(String servers, String topic, String groupId) { return new KafkaCanalConnector(servers, topic, null, groupId); } } diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java index c968f0f3..8b1ffc1d 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java @@ -1,6 +1,6 @@ package com.alibaba.otter.canal.kafka.client.running; -import com.alibaba.otter.canal.client.CanalConnector; +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.junit.Test; @@ -16,20 +16,19 @@ public class ClientRunningTest extends AbstractKafkaTest { public void testKafkaConsumer() { final ExecutorService executor = Executors.newFixedThreadPool(1); - final CanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); - kafkaCanalConnector.connect(); + final KafkaCanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); kafkaCanalConnector.subscribe(); executor.submit(new Runnable() { @Override public void run() { while (running) { - Message message = kafkaCanalConnector.getWithoutAck(1); + Message message = kafkaCanalConnector.getWithoutAck(); if (message != null) { System.out.println(message); sleep(40000); } - kafkaCanalConnector.ack(1); + kafkaCanalConnector.ack(); } } }); diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index a869321c..3861fe58 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -5,7 +5,7 @@ lingerMs: 1 bufferMemory: 33554432 topics: - - topic: expTest + - topic: example partition: canalDestination: example # - topic: example2 From b96e4dd23a693f352bbce5000e1facf85ed3a40d Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 11:13:35 +0800 Subject: [PATCH 12/27] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../kafka/client/KafkaCanalConnector.java | 10 +- .../client/running/AbstractKafkaTest.java | 2 +- .../client/running/ClientRunningTest.java | 30 ++++-- .../running/KafkaCanalClientExample.java | 92 +++++++++++++++++++ 4 files changed, 117 insertions(+), 17 deletions(-) create mode 100644 kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java index e3a68934..98200c06 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java @@ -1,7 +1,6 @@ package com.alibaba.otter.canal.kafka.client; import com.alibaba.otter.canal.protocol.Message; -import com.alibaba.otter.canal.protocol.exception.CanalClientException; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; @@ -73,16 +72,13 @@ public class KafkaCanalConnector { } /** - * 订阅topic，不能重复订阅 + * 订阅topic */ public void subscribe() { try { - if (kafkaConsumer != null) { - throw new CanalClientException("resubscribe error"); + if (kafkaConsumer == null) { + kafkaConsumer = new KafkaConsumer(properties); } - - kafkaConsumer = new KafkaConsumer(properties); - if (partition == null) { kafkaConsumer.subscribe(Collections.singletonList(topic)); } else { diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java index 2a9ce466..df03bc5c 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java @@ -6,7 +6,7 @@ public class AbstractKafkaTest { protected String topic = "example"; protected Integer partition = null; - protected String groupId = "g1"; + protected String groupId = "g2"; protected String servers = "slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667"; public void sleep(long time) { diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java index 8b1ffc1d..4ff3a159 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java @@ -3,12 +3,17 @@ package com.alibaba.otter.canal.kafka.client.running; 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.common.errors.WakeupException; import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; public class ClientRunningTest extends AbstractKafkaTest { + private Logger logger = LoggerFactory.getLogger(ClientRunningTest.class); private boolean running = true; @@ -16,26 +21,33 @@ public class ClientRunningTest extends AbstractKafkaTest { public void testKafkaConsumer() { final ExecutorService executor = Executors.newFixedThreadPool(1); - final KafkaCanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); - kafkaCanalConnector.subscribe(); + final KafkaCanalConnector connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); + connector.subscribe(); executor.submit(new Runnable() { @Override public void run() { while (running) { - Message message = kafkaCanalConnector.getWithoutAck(); - if (message != null) { - System.out.println(message); - sleep(40000); + try { + Message message = connector.getWithoutAck(3L, TimeUnit.SECONDS); + if (message != null) { + System.out.println(message); + } + connector.ack(); + } catch (WakeupException e) { + //ignore } - kafkaCanalConnector.ack(); } } }); - sleep(120000); + sleep(60000); + connector.disconnect(); running = false; - kafkaCanalConnector.disconnect(); + executor.shutdown(); + logger.info("shutdown completed"); + + sleep(10000); } } diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java new file mode 100644 index 00000000..ca304a27 --- /dev/null +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java @@ -0,0 +1,92 @@ +package com.alibaba.otter.canal.kafka.client.running; + +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.common.errors.WakeupException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.util.Assert; + +public class KafkaCanalClientExample { + protected final static Logger logger = LoggerFactory.getLogger(KafkaCanalClientExample.class); + + private KafkaCanalConnector connector; + + private volatile boolean running = false; + + private Thread thread = null; + + private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { + public void uncaughtException(Thread t, Throwable e) { + logger.error("parse events has an error", e); + } + }; + + public KafkaCanalClientExample(String servers, String topic, Integer partition, String groupId) { + connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); + } + + public void start() { + Assert.notNull(connector, "connector is null"); + thread = new Thread(new Runnable() { + + public void run() { + process(); + } + }); + + thread.setUncaughtExceptionHandler(handler); + running = true; + } + + public void stop() { + if (!running) { + return; + } + connector.disconnect(); + running = false; + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + } + + private void process() { + while (running) { + try { + connector.subscribe(); + while (running) { + try { + Message message = connector.getWithoutAck(); //获取message + long batchId = message.getId(); + int size = message.getEntries().size(); + if (batchId == -1 || size == 0) { + // try { + // Thread.sleep(1000); + // } catch (InterruptedException e) { + // } + } else { + // printSummary(message, batchId, size); + // printEntry(message.getEntries()); + logger.info(message.toString()); + } + + connector.ack(); // 提交确认 + } catch (WakeupException e) { + //ignore + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } catch (WakeupException e) { + //ignore + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } +} \ No newline at end of file From efb7b32694c9f097dd4d76d5a463d45402239a7d Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 13:36:28 +0800 Subject: [PATCH 13/27] =?UTF-8?q?kafka=20producer=20=E4=BF=AE=E6=94=B9?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=EF=BC=8C=E4=B8=80=E4=B8=AAdestination?= =?UTF-8?q?=E5=8F=AF=E4=BB=A5=E5=AF=B9=E5=BA=94=E5=A4=9A=E4=B8=AAkafka=20t?= =?UTF-8?q?opic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../kafka/producer/CanalKafkaStarter.java | 39 +++++---- .../canal/kafka/producer/KafkaProperties.java | 82 ++++++++++++++++--- kafka/src/main/resources/kafka.yml | 15 ++-- 3 files changed, 102 insertions(+), 34 deletions(-) diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index 3ef05489..5a166a69 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.kafka.producer; 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.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; @@ -31,13 +32,15 @@ public class CanalKafkaStarter { private static CanalKafkaProducer canalKafkaProducer; + private static KafkaProperties kafkaProperties; + public static void init() { try { logger.info("## load kafka configurations"); String conf = System.getProperty("kafka.conf", "classpath:kafka.yml"); - KafkaProperties kafkaProperties; + if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); kafkaProperties = new Yaml().loadAs(CanalKafkaStarter.class.getClassLoader().getResourceAsStream(conf), KafkaProperties.class); @@ -50,16 +53,16 @@ public class CanalKafkaStarter { canalKafkaProducer.init(kafkaProperties); //对应每个instance启动一个worker线程 - List topics = kafkaProperties.getTopics(); + List destinations = kafkaProperties.getCanalDestinations(); - executorService = Executors.newFixedThreadPool(topics.size()); + executorService = Executors.newFixedThreadPool(destinations.size()); logger.info("## start the kafka workers."); - for (final Topic topic : topics) { + for (final CanalDestination destination : destinations) { executorService.execute(new Runnable() { @Override public void run() { - worker(topic); + worker(destination); } }); } @@ -88,12 +91,12 @@ public class CanalKafkaStarter { } - private static void worker(Topic topic) { + private static void worker(CanalDestination destination) { while (!running) ; while (!CanalServerStarter.isRunning()) ; //等待server启动完成 - logger.info("## start the canal consumer: {}.", topic.getCanalDestination()); + logger.info("## start the canal consumer: {}.", destination.getCanalDestination()); CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); - ClientIdentity clientIdentity = new ClientIdentity(topic.getCanalDestination(), (short) 1001, ""); + ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); while (running) { try { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { @@ -105,21 +108,23 @@ public class CanalKafkaStarter { continue; } server.subscribe(clientIdentity); - logger.info("## the canal consumer {} is running now ......", topic.getCanalDestination()); + logger.info("## the canal consumer {} is running now ......", destination.getCanalDestination()); while (running) { - Message message = server.getWithoutAck(clientIdentity, 5 * 1024); // 获取指定数量的数据 + Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据 long batchId = message.getId(); try { int size = message.getEntries().size(); - if (batchId == -1 || size == 0) { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - //ignore + if (batchId != -1 && size != 0) { + if (!StringUtils.isEmpty(destination.getTopic())) { + Topic topic = new Topic(); + topic.setTopic(destination.getTopic()); + topic.setPartition(destination.getPartition()); + destination.getTopics().add(topic); + } + for (Topic topic : destination.getTopics()) { + canalKafkaProducer.send(topic, message); //发送message到所有topic } - } else { - canalKafkaProducer.send(topic, message); } if (batchId != -1) { diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java index f7362a78..653ad6b6 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java @@ -1,7 +1,9 @@ package com.alibaba.otter.canal.kafka.producer; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; /** * kafka 配置项 @@ -16,12 +18,23 @@ public class KafkaProperties { private int lingerMs = 1; private long bufferMemory = 33554432L; - private List topics = new ArrayList(); + private int canalBatchSize = 5; - public static class Topic { + private List canalDestinations = new ArrayList(); + + public static class CanalDestination { + private String canalDestination; private String topic; private Integer partition; - private String canalDestination; + private Set topics = new HashSet(); + + public String getCanalDestination() { + return canalDestination; + } + + public void setCanalDestination(String canalDestination) { + this.canalDestination = canalDestination; + } public String getTopic() { return topic; @@ -39,12 +52,51 @@ public class KafkaProperties { this.partition = partition; } - public String getCanalDestination() { - return canalDestination; + public Set getTopics() { + return topics; } - public void setCanalDestination(String canalDestination) { - this.canalDestination = canalDestination; + public void setTopics(Set topics) { + this.topics = topics; + } + } + + public static class Topic { + private String topic; + private Integer partition; + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public Integer getPartition() { + return partition; + } + + public void setPartition(Integer partition) { + this.partition = partition; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + Topic topic1 = (Topic) o; + + if (topic != null ? !topic.equals(topic1.topic) : topic1.topic != null) return false; + return partition != null ? partition.equals(topic1.partition) : topic1.partition == null; + } + + @Override + public int hashCode() { + int result = topic != null ? topic.hashCode() : 0; + result = 31 * result + (partition != null ? partition.hashCode() : 0); + return result; } } @@ -88,11 +140,19 @@ public class KafkaProperties { this.bufferMemory = bufferMemory; } - public List getTopics() { - return topics; + public int getCanalBatchSize() { + return canalBatchSize; } - public void setTopics(List topics) { - this.topics = topics; + public void setCanalBatchSize(int canalBatchSize) { + this.canalBatchSize = canalBatchSize; + } + + public List getCanalDestinations() { + return canalDestinations; + } + + public void setCanalDestinations(List canalDestinations) { + this.canalDestinations = canalDestinations; } } diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index 3861fe58..b46f123c 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -3,13 +3,16 @@ retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 +# canal的批次大小，单位 k +canalBatchSize: 5 -topics: - - topic: example +canalDestinations: + - canalDestination: example + topic: example partition: - canalDestination: example -# - topic: example2 -# partition: -# canalDestination: example + # 一个destination可以对应多个topic +# topics: +# - topics: example +# partition: From ec072a45d853723e0eb35496c49650a10163057c Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 13:49:15 +0800 Subject: [PATCH 14/27] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF=20exa?= =?UTF-8?q?mple?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/running/AbstractKafkaTest.java | 10 ++--- .../running/KafkaCanalClientExample.java | 39 +++++++++++++++++-- 2 files changed, 41 insertions(+), 8 deletions(-) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java index df03bc5c..c47e2d8e 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java @@ -2,12 +2,12 @@ package com.alibaba.otter.canal.kafka.client.running; import org.junit.Assert; -public class AbstractKafkaTest { +public abstract class AbstractKafkaTest { - protected String topic = "example"; - protected Integer partition = null; - protected String groupId = "g2"; - protected String servers = "slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667"; + public static String topic = "example"; + public static Integer partition = null; + public static String groupId = "g1"; + public static String servers = "slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667"; public void sleep(long time) { try { diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java index ca304a27..0f155c35 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java @@ -8,12 +8,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.Assert; +import java.util.concurrent.TimeUnit; + public class KafkaCanalClientExample { protected final static Logger logger = LoggerFactory.getLogger(KafkaCanalClientExample.class); private KafkaCanalConnector connector; - private volatile boolean running = false; + private static volatile boolean running = false; private Thread thread = null; @@ -27,6 +29,34 @@ public class KafkaCanalClientExample { connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); } + public static void main(String[] args) { + try { + final KafkaCanalClientExample kafkaCanalClientExample = new KafkaCanalClientExample(AbstractKafkaTest.servers, + AbstractKafkaTest.topic, AbstractKafkaTest.partition, AbstractKafkaTest.groupId); + logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId); + kafkaCanalClientExample.start(); + logger.info("## the canal kafka consumer is running now ......"); + Runtime.getRuntime().addShutdownHook(new Thread() { + + public void run() { + try { + logger.info("## stop the kafka consumer"); + kafkaCanalClientExample.stop(); + } catch (Throwable e) { + logger.warn("##something goes wrong when stopping kafka consumer:", e); + } finally { + logger.info("## kafka consumer is down."); + } + } + + }); + while (running) ; + } catch (Throwable e) { + logger.error("## Something goes wrong when starting up the kafka consumer:", e); + System.exit(0); + } + } + public void start() { Assert.notNull(connector, "connector is null"); thread = new Thread(new Runnable() { @@ -35,8 +65,8 @@ public class KafkaCanalClientExample { process(); } }); - thread.setUncaughtExceptionHandler(handler); + thread.start(); running = true; } @@ -61,7 +91,10 @@ public class KafkaCanalClientExample { connector.subscribe(); while (running) { try { - Message message = connector.getWithoutAck(); //获取message + Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); //获取message + if (message == null) { + continue; + } long batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { From 9964bb75b19fdbe7ea0e004d38ee6b2735b839c0 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 14:01:30 +0800 Subject: [PATCH 15/27] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF=20exa?= =?UTF-8?q?mple=20=E5=AE=8C=E5=96=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../running/KafkaCanalClientExample.java | 64 ++++++++++--------- 1 file changed, 33 insertions(+), 31 deletions(-) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java index 0f155c35..4527d847 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java @@ -86,40 +86,42 @@ public class KafkaCanalClientExample { } private void process() { - while (running) { - try { - connector.subscribe(); - while (running) { - try { - Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); //获取message - if (message == null) { - continue; - } - long batchId = message.getId(); - int size = message.getEntries().size(); - if (batchId == -1 || size == 0) { - // try { - // Thread.sleep(1000); - // } catch (InterruptedException e) { - // } - } else { - // printSummary(message, batchId, size); - // printEntry(message.getEntries()); - logger.info(message.toString()); - } - - connector.ack(); // 提交确认 - } catch (WakeupException e) { - //ignore - } catch (Exception e) { - logger.error(e.getMessage(), e); + try { + connector.subscribe(); + while (running) { + try { + Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); //获取message + if (message == null) { + continue; } + long batchId = message.getId(); + int size = message.getEntries().size(); + if (batchId == -1 || size == 0) { + // try { + // Thread.sleep(1000); + // } catch (InterruptedException e) { + // } + } else { + // printSummary(message, batchId, size); + // printEntry(message.getEntries()); + logger.info(message.toString()); + } + + connector.ack(); // 提交确认 + } catch (WakeupException e) { + try { + Thread.sleep(500); //延时确保running状态的改变 + } catch (InterruptedException ie) { + //ignore + } + } catch (Exception e) { + logger.error(e.getMessage(), e); } - } catch (WakeupException e) { - //ignore - } catch (Exception e) { - logger.error(e.getMessage(), e); } + } catch (WakeupException e) { + //ignore + } catch (Exception e) { + logger.error(e.getMessage(), e); } } } \ No newline at end of file From 345e22bd2ca9475a8ac89d04e22e0574300865f3 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 14:16:37 +0800 Subject: [PATCH 16/27] =?UTF-8?q?kafka=20=E8=AE=BE=E7=BD=AE=E7=8B=AC?= =?UTF-8?q?=E7=AB=8Blogback.xml?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/src/main/assembly/dev.xml | 3 + kafka/src/main/assembly/release.xml | 3 + .../kafka/producer/CanalKafkaProducer.java | 1 + kafka/src/main/resources/logback.xml | 86 +++++++++++++++++++ 4 files changed, 93 insertions(+) create mode 100644 kafka/src/main/resources/logback.xml diff --git a/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml index df9eca81..2c3e3fff 100644 --- a/kafka/src/main/assembly/dev.xml +++ b/kafka/src/main/assembly/dev.xml @@ -34,6 +34,9 @@ **/* + + logback.xml + ./src/main/resources diff --git a/kafka/src/main/assembly/release.xml b/kafka/src/main/assembly/release.xml index d79ecf3c..0b103620 100644 --- a/kafka/src/main/assembly/release.xml +++ b/kafka/src/main/assembly/release.xml @@ -34,6 +34,9 @@ **/* + + logback.xml + ./src/main/resources diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java index 21527291..fdd522aa 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -69,5 +69,6 @@ public class CanalKafkaProducer { record = new ProducerRecord(topic.getTopic(), message); } producer.send(record); + logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); } } diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml new file mode 100644 index 00000000..14c6d89e --- /dev/null +++ b/kafka/src/main/resources/logback.xml @@ -0,0 +1,86 @@ + + + + + %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 From 4bd2a79905fd71811b63a0199472dc5e37169039 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 14:17:07 +0800 Subject: [PATCH 17/27] =?UTF-8?q?kafka=20pom=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/pom.xml | 1 - 1 file changed, 1 deletion(-) diff --git a/kafka/pom.xml b/kafka/pom.xml index 68d6d72e..a4a7b13a 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -53,7 +53,6 @@ **/canal.properties **/spring/** **/example/** - **/example2/** **/kafka.yml From cb6c7ef3e8b468277a7b52e6caefce03e456617a Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 14:24:32 +0800 Subject: [PATCH 18/27] =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/kafka/client/KafkaCanalConnectors.java | 6 ++++++ .../canal/kafka/client/MessageDeserializer.java | 6 ++++++ .../kafka/client/running/AbstractKafkaTest.java | 6 ++++++ ...ntExample.java => CanalKafkaClientExample.java} | 14 ++++++++++---- ...unningTest.java => KafkaClientRunningTest.java} | 10 ++++++++-- .../canal/kafka/producer/MessageSerializer.java | 2 +- 6 files changed, 37 insertions(+), 7 deletions(-) rename kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/{KafkaCanalClientExample.java => CanalKafkaClientExample.java} (92%) rename kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/{ClientRunningTest.java => KafkaClientRunningTest.java} (85%) diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java index eb5f248c..26c0c5ae 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java @@ -1,5 +1,11 @@ package com.alibaba.otter.canal.kafka.client; +/** + * canal kafka connectors创建工具类 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ public class KafkaCanalConnectors { /** * 创建kafka客户端链接 diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java index c321d076..c6a7ee83 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java @@ -11,6 +11,12 @@ import org.slf4j.LoggerFactory; import java.util.Map; +/** + * Kafka Message类的反序列化 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ public class MessageDeserializer implements Deserializer { private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class); diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java index c47e2d8e..f1e52be4 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java @@ -2,6 +2,12 @@ package com.alibaba.otter.canal.kafka.client.running; import org.junit.Assert; +/** + * Kafka 测试基类 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ public abstract class AbstractKafkaTest { public static String topic = "example"; diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java similarity index 92% rename from kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java rename to kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java index 4527d847..5a7fe4f0 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java @@ -10,8 +10,14 @@ import org.springframework.util.Assert; import java.util.concurrent.TimeUnit; -public class KafkaCanalClientExample { - protected final static Logger logger = LoggerFactory.getLogger(KafkaCanalClientExample.class); +/** + * Kafka client example + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ +public class CanalKafkaClientExample { + protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaClientExample.class); private KafkaCanalConnector connector; @@ -25,13 +31,13 @@ public class KafkaCanalClientExample { } }; - public KafkaCanalClientExample(String servers, String topic, Integer partition, String groupId) { + public CanalKafkaClientExample(String servers, String topic, Integer partition, String groupId) { connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); } public static void main(String[] args) { try { - final KafkaCanalClientExample kafkaCanalClientExample = new KafkaCanalClientExample(AbstractKafkaTest.servers, + final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.servers, AbstractKafkaTest.topic, AbstractKafkaTest.partition, AbstractKafkaTest.groupId); logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId); kafkaCanalClientExample.start(); diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java similarity index 85% rename from kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java rename to kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java index 4ff3a159..3260c657 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java @@ -12,8 +12,14 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -public class ClientRunningTest extends AbstractKafkaTest { - private Logger logger = LoggerFactory.getLogger(ClientRunningTest.class); +/** + * Kafka consumer获取Message的测试例子 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ +public class KafkaClientRunningTest extends AbstractKafkaTest { + private Logger logger = LoggerFactory.getLogger(KafkaClientRunningTest.class); private boolean running = true; diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java index d1a9577a..c74eb9df 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java @@ -10,7 +10,7 @@ import org.springframework.util.CollectionUtils; import java.util.Map; /** - * Message类的序列化Kafka实现 + * Kafka Message类的序列化 * * @author machengyuan 2018-6-11 下午05:30:49 * @version 1.0.0 From 1b6efdf46399af7e34b891ce97b4c367513aebac Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 15:25:43 +0800 Subject: [PATCH 19/27] =?UTF-8?q?=E4=BF=AE=E6=94=B9kafka=20consumer?= =?UTF-8?q?=E5=85=B3=E9=97=AD=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../kafka/client/KafkaCanalConnector.java | 64 +++++-------------- .../running/CanalKafkaClientExample.java | 16 ++--- 2 files changed, 24 insertions(+), 56 deletions(-) diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java index 98200c06..763346b9 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java @@ -4,11 +4,9 @@ import com.alibaba.otter.canal.protocol.Message; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import java.util.Collections; -import java.util.ConcurrentModificationException; import java.util.Properties; import java.util.concurrent.TimeUnit; @@ -59,35 +57,24 @@ public class KafkaCanalConnector { } /** - * 释放链接 + * 关闭链接 */ - public void disconnect() { - if (kafkaConsumer != null) { - try { - kafkaConsumer.close(); - } catch (ConcurrentModificationException e) { - kafkaConsumer.wakeup(); //通过wakeup异常间接关闭consumer - } - } + public void close() { + kafkaConsumer.close(); } /** * 订阅topic */ public void subscribe() { - try { - if (kafkaConsumer == null) { - kafkaConsumer = new KafkaConsumer(properties); - } - if (partition == null) { - kafkaConsumer.subscribe(Collections.singletonList(topic)); - } else { - kafkaConsumer.subscribe(Collections.singletonList(topic)); - TopicPartition topicPartition = new TopicPartition(topic, partition); - kafkaConsumer.assign(Collections.singletonList(topicPartition)); - } - } catch (WakeupException e) { - closeByWakeupException(e); + if (kafkaConsumer == null) { + kafkaConsumer = new KafkaConsumer(properties); + } + if (partition == null) { + kafkaConsumer.subscribe(Collections.singletonList(topic)); + } else { + TopicPartition topicPartition = new TopicPartition(topic, partition); + kafkaConsumer.assign(Collections.singletonList(topicPartition)); } } @@ -95,11 +82,7 @@ public class KafkaCanalConnector { * 取消订阅 */ public void unsubscribe() { - try { - kafkaConsumer.unsubscribe(); - } catch (WakeupException e) { - closeByWakeupException(e); - } + kafkaConsumer.unsubscribe(); } /** @@ -127,15 +110,11 @@ public class KafkaCanalConnector { * @return */ public Message getWithoutAck(Long timeout, TimeUnit unit) { - try { - ConsumerRecords records = - kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据 + ConsumerRecords records = + kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据 - if (!records.isEmpty()) { - return records.iterator().next().value(); - } - } catch (WakeupException e) { - closeByWakeupException(e); + if (!records.isEmpty()) { + return records.iterator().next().value(); } return null; } @@ -144,15 +123,6 @@ public class KafkaCanalConnector { * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败 */ public void ack() { - try { - kafkaConsumer.commitSync(); - } catch (WakeupException e) { - closeByWakeupException(e); - } - } - - private void closeByWakeupException(WakeupException e) { - kafkaConsumer.close(); - throw e; + kafkaConsumer.commitSync(); } } diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java index 5a7fe4f0..a7c6b78c 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java @@ -80,7 +80,6 @@ public class CanalKafkaClientExample { if (!running) { return; } - connector.disconnect(); running = false; if (thread != null) { try { @@ -114,20 +113,19 @@ public class CanalKafkaClientExample { } connector.ack(); // 提交确认 - } catch (WakeupException e) { - try { - Thread.sleep(500); //延时确保running状态的改变 - } catch (InterruptedException ie) { - //ignore - } } catch (Exception e) { logger.error(e.getMessage(), e); } } - } catch (WakeupException e) { - //ignore } catch (Exception e) { logger.error(e.getMessage(), e); } + + try { + connector.unsubscribe(); + } catch (WakeupException e) { + // No-op. Continue process + } + connector.close(); } } \ No newline at end of file From 31cbdb1824dc5ea36b425ad3639b0393b6c70306 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 15:36:21 +0800 Subject: [PATCH 20/27] =?UTF-8?q?kafka=20consumer=20close=E6=96=B9?= =?UTF-8?q?=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/kafka/client/running/KafkaClientRunningTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java index 3260c657..b68d9ce7 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java @@ -48,8 +48,10 @@ public class KafkaClientRunningTest extends AbstractKafkaTest { }); sleep(60000); - connector.disconnect(); running = false; + connector.unsubscribe(); + connector.close(); + executor.shutdown(); logger.info("shutdown completed"); From 4e41ebf746822ed098e9d6b43a08b807a9d84df1 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 15:38:24 +0800 Subject: [PATCH 21/27] =?UTF-8?q?kafka=20consumer=20close=E6=96=B9?= =?UTF-8?q?=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/kafka/client/running/KafkaClientRunningTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java index b68d9ce7..39752364 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java @@ -44,14 +44,13 @@ public class KafkaClientRunningTest extends AbstractKafkaTest { //ignore } } + connector.unsubscribe(); + connector.close(); } }); sleep(60000); running = false; - connector.unsubscribe(); - connector.close(); - executor.shutdown(); logger.info("shutdown completed"); From 19f4c5ae1bd88522048cf32dcb598c48ab2e0b58 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 15:40:49 +0800 Subject: [PATCH 22/27] =?UTF-8?q?kafka=20consumer=20=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E7=B1=BB=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/kafka/client/running/KafkaClientRunningTest.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java index 39752364..9d462554 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java @@ -28,11 +28,11 @@ public class KafkaClientRunningTest extends AbstractKafkaTest { final ExecutorService executor = Executors.newFixedThreadPool(1); final KafkaCanalConnector connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); - connector.subscribe(); executor.submit(new Runnable() { @Override public void run() { + connector.subscribe(); while (running) { try { Message message = connector.getWithoutAck(3L, TimeUnit.SECONDS); @@ -53,8 +53,6 @@ public class KafkaClientRunningTest extends AbstractKafkaTest { running = false; executor.shutdown(); logger.info("shutdown completed"); - - sleep(10000); } } From cc5eb1075ba82eeef4fb4e88b8a8235c062ab272 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 15:51:28 +0800 Subject: [PATCH 23/27] =?UTF-8?q?kafka=20consumer=20example=20=E4=BF=AE?= =?UTF-8?q?=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/kafka/client/running/CanalKafkaClientExample.java | 1 + 1 file changed, 1 insertion(+) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java index a7c6b78c..f7c22cd4 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java @@ -91,6 +91,7 @@ public class CanalKafkaClientExample { } private void process() { + while (!running) ; try { connector.subscribe(); while (running) { From df668a6f72e2b1a64c06ee06f49ddf5d6541d2d8 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 17:20:16 +0800 Subject: [PATCH 24/27] =?UTF-8?q?kafka=20=E4=BF=AE=E6=94=B9=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E8=BE=93=E5=87=BA=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/src/main/resources/logback.xml | 1 - 1 file changed, 1 deletion(-) diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml index 14c6d89e..34e7b1e5 100644 --- a/kafka/src/main/resources/logback.xml +++ b/kafka/src/main/resources/logback.xml @@ -76,7 +76,6 @@ - From d014167226e145e1569552d8b251e9402f1de8f8 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 17:57:49 +0800 Subject: [PATCH 25/27] =?UTF-8?q?kafka=20consumer=20=E9=BB=98=E8=AE=A4sess?= =?UTF-8?q?ion=20timeout=20=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/otter/canal/kafka/client/KafkaCanalConnector.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java index 763346b9..41e82562 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java @@ -37,8 +37,8 @@ public class KafkaCanalConnector { properties.put("enable.auto.commit", false); properties.put("auto.commit.interval.ms", "1000"); properties.put("auto.offset.reset", "latest"); //earliest //如果没有offset则从最后的offset开始读 - properties.put("request.timeout.ms", "360000"); //必须大于session.timeout.ms的设置 - properties.put("session.timeout.ms", "300000"); //默认为5分钟 + properties.put("request.timeout.ms", "40000"); //必须大于session.timeout.ms的设置 + properties.put("session.timeout.ms", "30000"); //默认为30秒 properties.put("max.poll.records", "1"); //所以一次只取一条数据 properties.put("key.deserializer", StringDeserializer.class.getName()); properties.put("value.deserializer", MessageDeserializer.class.getName()); From 593902d349518d27e06562fa755e43b111bcdf1c Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 18:25:27 +0800 Subject: [PATCH 26/27] =?UTF-8?q?=E4=BF=AE=E6=94=B9example=E7=9A=84?= =?UTF-8?q?=E4=B8=80=E4=BA=9Bbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/otter/canal/example/AbstractCanalClientTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java index 33ea6550..183d65ce 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java @@ -38,7 +38,6 @@ public class AbstractCanalClientTest { protected static final String SEP = SystemUtils.LINE_SEPARATOR; protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss"; protected volatile boolean running = false; - private volatile boolean waiting = true; protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { public void uncaughtException(Thread t, Throwable e) { @@ -97,6 +96,7 @@ public class AbstractCanalClientTest { return; } connector.stopRunning(); + running = false; if (thread != null) { try { thread.join(); @@ -115,7 +115,6 @@ public class AbstractCanalClientTest { MDC.put("destination", destination); connector.connect(); connector.subscribe(); - waiting = false; while (running) { Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId(); From e23a19cd8555b5ccd00e9f7cf317a8541bf5b72a Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 18:29:15 +0800 Subject: [PATCH 27/27] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/client/running/AbstractKafkaTest.java | 2 +- kafka/src/main/resources/kafka.yml | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java index f1e52be4..4d621e10 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java @@ -13,7 +13,7 @@ public abstract class AbstractKafkaTest { public static String topic = "example"; public static Integer partition = null; public static String groupId = "g1"; - public static String servers = "slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667"; + public static String servers = "slave1:6667,slave2:6667,slave3:6667"; public void sleep(long time) { try { diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index b46f123c..ab214319 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -1,10 +1,10 @@ -servers: slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667 +servers: slave1:6667,slave2:6667,slave3:6667 retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 # canal的批次大小，单位 k -canalBatchSize: 5 +canalBatchSize: 50 canalDestinations: - canalDestination: example