From 1b11aa1d2190d469579475dc43f648475f2ed51b Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 6 Dec 2018 10:31:13 +0800 Subject: [PATCH 01/15] =?UTF-8?q?fixed=20issue=20#1220,=20=E8=A7=A3?= =?UTF-8?q?=E5=86=B3=E4=B8=80=E4=B8=8Bno=20alive=20canal=20server=E9=87=8D?= =?UTF-8?q?=E8=AF=95=E8=BF=87=E4=BA=8E=E9=A2=91=E7=B9=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../impl/ClusterNodeAccessStrategy.java | 5 ++-- .../client/impl/ServerNotFoundException.java | 29 +++++++++++++++++++ .../impl/running/ClientRunningMonitor.java | 10 +++++++ 3 files changed, 42 insertions(+), 2 deletions(-) create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/impl/ServerNotFoundException.java diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterNodeAccessStrategy.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterNodeAccessStrategy.java index 6461a974..269d99dc 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterNodeAccessStrategy.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterNodeAccessStrategy.java @@ -15,7 +15,6 @@ import com.alibaba.otter.canal.common.utils.JsonUtils; import com.alibaba.otter.canal.common.zookeeper.ZkClientx; import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData; -import com.alibaba.otter.canal.protocol.exception.CanalClientException; /** * 集群模式的调度策略 @@ -25,6 +24,7 @@ import com.alibaba.otter.canal.protocol.exception.CanalClientException; */ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy { + private String destination; private IZkChildListener childListener; // 监听所有的服务器列表 private IZkDataListener dataListener; // 监听当前的工作节点 private ZkClientx zkClient; @@ -32,6 +32,7 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy { private volatile InetSocketAddress runningAddress = null; public ClusterNodeAccessStrategy(String destination, ZkClientx zkClient){ + this.destination = destination; this.zkClient = zkClient; childListener = new IZkChildListener() { @@ -73,7 +74,7 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy { } else if (!currentAddress.isEmpty()) { // 如果不存在已经启动的服务，可能服务是一种lazy启动，随机选择一台触发服务器进行启动 return currentAddress.get(0);// 默认返回第一个节点，之前已经做过shuffle } else { - throw new CanalClientException("no alive canal server"); + throw new ServerNotFoundException("no alive canal server for " + destination); } } diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/ServerNotFoundException.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/ServerNotFoundException.java new file mode 100644 index 00000000..22f95046 --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/ServerNotFoundException.java @@ -0,0 +1,29 @@ +package com.alibaba.otter.canal.client.impl; + +import com.alibaba.otter.canal.protocol.exception.CanalClientException; + +public class ServerNotFoundException extends CanalClientException { + + private static final long serialVersionUID = -3471518241911601774L; + + public ServerNotFoundException(String errorCode, String errorDesc, Throwable cause){ + super(errorCode, errorDesc, cause); + } + + public ServerNotFoundException(String errorCode, String errorDesc){ + super(errorCode, errorDesc); + } + + public ServerNotFoundException(String errorCode, Throwable cause){ + super(errorCode, cause); + } + + public ServerNotFoundException(String errorCode){ + super(errorCode); + } + + public ServerNotFoundException(Throwable cause){ + super(cause); + } + +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java index 90a9ad71..9de7f4ee 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java @@ -16,6 +16,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; +import com.alibaba.otter.canal.client.impl.ServerNotFoundException; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.utils.BooleanMutex; import com.alibaba.otter.canal.common.utils.JsonUtils; @@ -141,6 +142,15 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle { logger.error(MessageFormat.format("There is an error when execute initRunning method, with destination [{0}].", destination), t); + + // fixed issue 1220, 针对server节点不工作避免死循环 + if (t instanceof ServerNotFoundException) { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + } + } + // 出现任何异常尝试release releaseRunning(); throw new CanalClientException("something goes wrong in initRunning method. ", t); From c045dc244edc81435d11a3dc1bacfee3f8a445a7 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 6 Dec 2018 10:50:12 +0800 Subject: [PATCH 02/15] optimizer filter --- .../filter/aviater/AviaterRegexFilter.java | 4 +++ .../mysql/AbstractMysqlEventParser.java | 36 ++++++++++++------- .../parse/inbound/mysql/MysqlConnection.java | 4 +-- .../inbound/mysql/dbsync/LogEventConvert.java | 2 ++ pom.xml | 2 ++ 5 files changed, 34 insertions(+), 14 deletions(-) diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java index b634c40a..026fe006 100644 --- a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java +++ b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java @@ -125,4 +125,8 @@ public class AviaterRegexFilter implements CanalEventFilter { return result; } + @Override + public String toString() { + return pattern; + } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index af792ec8..29a66fd2 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -11,6 +11,7 @@ import com.alibaba.otter.canal.parse.inbound.AbstractEventParser; import com.alibaba.otter.canal.parse.inbound.BinlogParser; import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBFactory; @@ -64,8 +65,29 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { super.setEventFilter(eventFilter); // 触发一下filter变更 - if (eventFilter != null && eventFilter instanceof AviaterRegexFilter && binlogParser instanceof LogEventConvert) { - ((LogEventConvert) binlogParser).setNameFilter((AviaterRegexFilter) eventFilter); + if (eventFilter != null && eventFilter instanceof AviaterRegexFilter) { + if (binlogParser instanceof LogEventConvert) { + ((LogEventConvert) binlogParser).setNameFilter((AviaterRegexFilter) eventFilter); + } + + if (tableMetaTSDB != null && tableMetaTSDB instanceof DatabaseTableMeta) { + ((DatabaseTableMeta) tableMetaTSDB).setFilter(eventFilter); + } + } + } + + public void setEventBlackFilter(CanalEventFilter eventBlackFilter) { + super.setEventBlackFilter(eventBlackFilter); + + // 触发一下filter变更 + if (eventBlackFilter != null && eventBlackFilter instanceof AviaterRegexFilter) { + if (binlogParser instanceof LogEventConvert) { + ((LogEventConvert) binlogParser).setNameBlackFilter((AviaterRegexFilter) eventBlackFilter); + } + + if (tableMetaTSDB != null && tableMetaTSDB instanceof DatabaseTableMeta) { + ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter); + } } } @@ -115,16 +137,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { super.stop(); } - public void setEventBlackFilter(CanalEventFilter eventBlackFilter) { - super.setEventBlackFilter(eventBlackFilter); - - // 触发一下filter变更 - if (eventBlackFilter != null && eventBlackFilter instanceof AviaterRegexFilter - && binlogParser instanceof LogEventConvert) { - ((LogEventConvert) binlogParser).setNameBlackFilter((AviaterRegexFilter) eventBlackFilter); - } - } - protected MultiStageCoprocessor buildMultiStageCoprocessor() { MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize, parallelThreadSize, diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 0ec5e7aa..7ab61ba5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -506,8 +506,8 @@ public class MysqlConnection implements ErosaConnection { ResultSetPacket rs = null; try { rs = query("select @@global.binlog_checksum"); - } catch (IOException e) { - throw new CanalParseException(e); + } catch (Throwable e) { + // ignore } List columnValues = rs.getFieldValues(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 2832ff34..3f811971 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -974,10 +974,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setNameFilter(AviaterRegexFilter nameFilter) { this.nameFilter = nameFilter; + logger.warn("--> init table filter : " + nameFilter.toString()); } public void setNameBlackFilter(AviaterRegexFilter nameBlackFilter) { this.nameBlackFilter = nameBlackFilter; + logger.warn("--> init table black filter : " + nameBlackFilter.toString()); } public void setTableMetaCache(TableMetaCache tableMetaCache) { diff --git a/pom.xml b/pom.xml index e5ba86c9..266b1921 100644 --- a/pom.xml +++ b/pom.xml @@ -381,6 +381,7 @@ + src/main/java src/test/java From ce956c980dc73dbee8cdc45e5b0663bcb0969d34 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 6 Dec 2018 18:48:17 +0800 Subject: [PATCH 03/15] add github template --- .github/issue_template.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .github/issue_template.md diff --git a/.github/issue_template.md b/.github/issue_template.md new file mode 100644 index 00000000..df5f2769 --- /dev/null +++ b/.github/issue_template.md @@ -0,0 +1,13 @@ +### environment + +* canal version +* mysql version + +### Issue Description + + +### Steps to reproduce + +### Expected behaviour + +### Actual behaviour From 5a99de1fa966ca1c6e6d1b829e59d46907507c6d Mon Sep 17 00:00:00 2001 From: tankilo Date: Fri, 7 Dec 2018 14:53:31 +0800 Subject: [PATCH 04/15] switch file format from DOS to UNIX --- deployer/src/main/bin/restart.sh | 5 -- deployer/src/main/bin/startup.sh | 104 ------------------------------- deployer/src/main/bin/stop.sh | 65 ------------------- 3 files changed, 174 deletions(-) delete mode 100644 deployer/src/main/bin/restart.sh delete mode 100644 deployer/src/main/bin/startup.sh delete mode 100644 deployer/src/main/bin/stop.sh diff --git a/deployer/src/main/bin/restart.sh b/deployer/src/main/bin/restart.sh deleted file mode 100644 index 63749ce5..00000000 --- a/deployer/src/main/bin/restart.sh +++ /dev/null @@ -1,5 +0,0 @@ -#!/bin/bash - -sh stop.sh - -sh startup.sh diff --git a/deployer/src/main/bin/startup.sh b/deployer/src/main/bin/startup.sh deleted file mode 100644 index 5655c713..00000000 --- a/deployer/src/main/bin/startup.sh +++ /dev/null @@ -1,104 +0,0 @@ -#!/bin/bash - -current_path=`pwd` -case "`uname`" in - Linux) - bin_abs_path=$(readlink -f $(dirname $0)) - ;; - *) - bin_abs_path=`cd $(dirname $0); pwd` - ;; -esac -base=${bin_abs_path}/.. -canal_conf=$base/conf/canal.properties -logback_configurationFile=$base/conf/logback.xml -export LANG=en_US.UTF-8 -export BASE=$base - -if [ -f $base/bin/canal.pid ] ; then - echo "found canal.pid , Please run stop.sh first ,then startup.sh" 2>&2 - exit 1 -fi - -if [ ! -d $base/logs/canal ] ; then - mkdir -p $base/logs/canal -fi - -## set java path -if [ -z "$JAVA" ] ; then - JAVA=$(which java) -fi - -ALIBABA_JAVA="/usr/alibaba/java/bin/java" -TAOBAO_JAVA="/opt/taobao/java/bin/java" -if [ -z "$JAVA" ]; then - if [ -f $ALIBABA_JAVA ] ; then - JAVA=$ALIBABA_JAVA - elif [ -f $TAOBAO_JAVA ] ; then - JAVA=$TAOBAO_JAVA - else - echo "Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH." 2>&2 - exit 1 - fi -fi - -case "$#" -in -0 ) - ;; -1 ) - var=$* - if [ -f $var ] ; then - canal_conf=$var - else - echo "THE PARAMETER IS NOT CORRECT.PLEASE CHECK AGAIN." - exit - fi;; -2 ) - var=$1 - if [ -f $var ] ; then - canal_conf=$var - else - if [ "$1" = "debug" ]; then - DEBUG_PORT=$2 - DEBUG_SUSPEND="n" - JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND" - fi - fi;; -* ) - echo "THE PARAMETERS MUST BE TWO OR LESS.PLEASE CHECK AGAIN." - exit;; -esac - -str=`file -L $JAVA | grep 64-bit` -if [ -n "$str" ]; then - JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -Xss256k -XX:-UseAdaptiveSizePolicy -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseFastAccessorMethods -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError" -else - JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m " -fi - -JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8" -CANAL_OPTS="-DappName=otter-canal -Dlogback.configurationFile=$logback_configurationFile -Dcanal.conf=$canal_conf" - -if [ -e $canal_conf -a -e $logback_configurationFile ] -then - - for i in $base/lib/*; - do CLASSPATH=$i:"$CLASSPATH"; - done - CLASSPATH="$base/conf:$CLASSPATH"; - - echo "cd to $bin_abs_path for workaround relative path" - cd $bin_abs_path - - echo LOG CONFIGURATION : $logback_configurationFile - echo canal conf : $canal_conf - echo CLASSPATH :$CLASSPATH - $JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.deployer.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/deployer/src/main/bin/stop.sh b/deployer/src/main/bin/stop.sh deleted file mode 100644 index f398749c..00000000 --- a/deployer/src/main/bin/stop.sh +++ /dev/null @@ -1,65 +0,0 @@ -#!/bin/bash - -cygwin=false; -linux=false; -case "`uname`" in - CYGWIN*) - cygwin=true - ;; - Linux*) - linux=true - ;; -esac - -get_pid() { - STR=$1 - PID=$2 - if $cygwin; then - JAVA_CMD="$JAVA_HOME\bin\java" - JAVA_CMD=`cygpath --path --unix $JAVA_CMD` - JAVA_PID=`ps |grep $JAVA_CMD |awk '{print $1}'` - else - if $linux; then - if [ ! -z "$PID" ]; then - JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` - else - JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'` - fi - else - if [ ! -z "$PID" ]; then - JAVA_PID=`ps aux |grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` - else - JAVA_PID=`ps aux |grep "$STR"|grep -v grep|awk '{print $2}'` - fi - fi - fi - echo $JAVA_PID; -} - -base=`dirname $0`/.. -pidfile=$base/bin/canal.pid -if [ ! -f "$pidfile" ];then - echo "canal is not running. exists" - exit -fi - -pid=`cat $pidfile` -if [ "$pid" == "" ] ; then - pid=`get_pid "appName=otter-canal"` -fi - -echo -e "`hostname`: stopping canal $pid ... " -kill $pid - -LOOPS=0 -while (true); -do - gpid=`get_pid "appName=otter-canal" "$pid"` - if [ "$gpid" == "" ] ; then - echo "Oook! cost:$LOOPS" - `rm $pidfile` - break; - fi - let LOOPS=LOOPS+1 - sleep 1 -done \ No newline at end of file From f0fc37c43bf02bc38ea3eeabfbc54f11ef70275f Mon Sep 17 00:00:00 2001 From: tankilo Date: Fri, 7 Dec 2018 14:54:37 +0800 Subject: [PATCH 05/15] switch file format from DOS to UNIX --- deployer/src/main/bin/restart.sh | 5 ++ deployer/src/main/bin/startup.sh | 104 +++++++++++++++++++++++++++++++ deployer/src/main/bin/stop.sh | 65 +++++++++++++++++++ 3 files changed, 174 insertions(+) create mode 100644 deployer/src/main/bin/restart.sh create mode 100644 deployer/src/main/bin/startup.sh create mode 100644 deployer/src/main/bin/stop.sh diff --git a/deployer/src/main/bin/restart.sh b/deployer/src/main/bin/restart.sh new file mode 100644 index 00000000..63749ce5 --- /dev/null +++ b/deployer/src/main/bin/restart.sh @@ -0,0 +1,5 @@ +#!/bin/bash + +sh stop.sh + +sh startup.sh diff --git a/deployer/src/main/bin/startup.sh b/deployer/src/main/bin/startup.sh new file mode 100644 index 00000000..5655c713 --- /dev/null +++ b/deployer/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.deployer.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/deployer/src/main/bin/stop.sh b/deployer/src/main/bin/stop.sh new file mode 100644 index 00000000..f398749c --- /dev/null +++ b/deployer/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 From 03ee9c7ecf8120b5fe8d47ed2c0b3b64dc376e01 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 00:09:49 +0800 Subject: [PATCH 06/15] =?UTF-8?q?=E6=A0=B9=E6=8D=AE=E4=B8=BB=E9=94=AEHash?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=AD=A3=E5=88=99=E8=A1=A8=E8=BE=BE=E5=BC=8F?= =?UTF-8?q?=E6=94=AF=E6=8C=81=EF=BC=8Cpk=E9=80=9A=E9=85=8D=E7=AC=A6?= =?UTF-8?q?=E6=8C=87=E5=AE=9A=E8=87=AA=E5=8A=A8=E8=8E=B7=E5=8F=96=E6=94=AF?= =?UTF-8?q?=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/instance/core/CanalMQConfig.java | 34 +--- .../otter/canal/protocol/FlatMessage.java | 168 ++++++++++++------ .../otter/canal/common/MQProperties.java | 15 +- .../otter/canal/server/CanalMQStarter.java | 2 +- 4 files changed, 123 insertions(+), 96 deletions(-) diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java index c275f150..53c50742 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java @@ -1,16 +1,11 @@ package com.alibaba.otter.canal.instance.core; -import java.util.LinkedHashMap; -import java.util.Map; - public class CanalMQConfig { - private String topic; - private Integer partition; - private Integer partitionsNum; - private String partitionHash; - - private volatile Map partitionHashProperties; + private String topic; + private Integer partition; + private Integer partitionsNum; + private String partitionHash; public String getTopic() { return topic; @@ -44,25 +39,4 @@ public class CanalMQConfig { this.partitionHash = partitionHash; } - public Map getPartitionHashProperties() { - if (partitionHashProperties == null) { - synchronized (CanalMQConfig.class) { - if (partitionHashProperties == null) { - if (partitionHash != null) { - partitionHashProperties = new LinkedHashMap<>(); - String[] items = partitionHash.split(","); - for (String item : items) { - int i = item.indexOf(":"); - if (i > -1) { - String dbTable = item.substring(0, i).trim(); - String pk = item.substring(i + 1).trim(); - partitionHashProperties.put(dbTable, pk); - } - } - } - } - } - } - return partitionHashProperties; - } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index 94c539cb..568d689a 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -1,12 +1,10 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; +import java.util.regex.Pattern; + +import org.apache.commons.lang.StringUtils; import com.google.protobuf.ByteString; @@ -33,8 +31,7 @@ public class FlatMessage implements Serializable { private List> data; private List> old; - public FlatMessage(){ - } + private transient Message message; // 所属message public FlatMessage(long id){ this.id = id; @@ -136,6 +133,14 @@ public class FlatMessage implements Serializable { this.es = es; } + public Message getMessage() { + return message; + } + + public void setMessage(Message message) { + this.message = message; + } + /** * 将Message转换为FlatMessage * @@ -187,6 +192,7 @@ public class FlatMessage implements Serializable { flatMessage.setEs(entry.getHeader().getExecuteTime()); flatMessage.setTs(System.currentTimeMillis()); flatMessage.setSql(rowChange.getSql()); + flatMessage.setMessage(message); if (!rowChange.getIsDdl()) { Map sqlType = new LinkedHashMap<>(); @@ -269,70 +275,116 @@ public class FlatMessage implements Serializable { * * @param flatMessage flatMessage * @param partitionsNum 分区数量 - * @param pkHashConfig hash映射 + * @param pkHashConfigs hash映射 * @return 拆分后的flatMessage数组 */ - public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, - Map pkHashConfig) { + public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs) { if (partitionsNum == null) { partitionsNum = 1; } FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; - String pk = pkHashConfig.get(flatMessage.getDatabase() + "." + flatMessage.getTable()); - if (pk == null || flatMessage.getIsDdl()) { + + if (flatMessage.getIsDdl()) { partitionMessages[0] = flatMessage; } else { if (flatMessage.getData() != null) { - int idx = 0; - for (Map row : flatMessage.getData()) { - Map o = null; - if (flatMessage.getOld() != null) { - o = flatMessage.getOld().get(idx); - } - String value; - // 如果old中有pk值说明主键有修改, 以旧的主键值hash为准 - if (o != null && o.containsKey(pk)) { - value = o.get(pk); - } else { - value = row.get(pk); - } - if (value == null) { - value = ""; - } - int hash = value.hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 - pkHash = Math.abs(pkHash); + String database = flatMessage.getDatabase(); + String table = flatMessage.getTable(); - FlatMessage flatMessageTmp = partitionMessages[pkHash]; - if (flatMessageTmp == null) { - flatMessageTmp = new FlatMessage(flatMessage.getId()); - partitionMessages[pkHash] = flatMessageTmp; - flatMessageTmp.setDatabase(flatMessage.getDatabase()); - flatMessageTmp.setTable(flatMessage.getTable()); - flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); - flatMessageTmp.setType(flatMessage.getType()); - flatMessageTmp.setSql(flatMessage.getSql()); - flatMessageTmp.setSqlType(flatMessage.getSqlType()); - flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); - flatMessageTmp.setEs(flatMessage.getEs()); - flatMessageTmp.setTs(flatMessage.getTs()); + String pk = null; + boolean isMatch = false; + + String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, ","); + for (String pkHashConfig : pkHashConfigArray) { + int i = pkHashConfig.lastIndexOf("."); + if (!pkHashConfig.endsWith(".$pk$")) { + // 如果指定了主键 + pk = pkHashConfig.substring(i + 1); } - List> data = flatMessageTmp.getData(); - if (data == null) { - data = new ArrayList<>(); - flatMessageTmp.setData(data); + pkHashConfig = pkHashConfig.substring(0, i); + isMatch = Pattern.matches(pkHashConfig, database + "." + table); + if (isMatch) { + break; } - data.add(row); - if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { - List> old = flatMessageTmp.getOld(); - if (old == null) { - old = new ArrayList<>(); - flatMessageTmp.setOld(old); + } + + if (!isMatch) { + // 如果都没有匹配，发送到第一个分区 + partitionMessages[0] = flatMessage; + } else { + if (pk == null) { + // 如果未指定主键(通配符主键)，从原生message中取主键字段 + CanalEntry.Entry entry = flatMessage.getMessage().getEntries().get(0); + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pk = column.getName(); + break; + } + } + } + if (pk == null || !flatMessage.getData().get(0).containsKey(pk)) { + // 如果都没有匹配的主键，发送到第一个分区 + partitionMessages[0] = flatMessage; + } else { + int idx = 0; + for (Map row : flatMessage.getData()) { + Map o = null; + if (flatMessage.getOld() != null) { + o = flatMessage.getOld().get(idx); + } + String value; + // 如果old中有pk值说明主键有修改, 以旧的主键值hash为准 + if (o != null && o.containsKey(pk)) { + value = o.get(pk); + } else { + value = row.get(pk); + } + if (value == null) { + value = ""; + } + int hash = value.hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 + pkHash = Math.abs(pkHash); + + FlatMessage flatMessageTmp = partitionMessages[pkHash]; + if (flatMessageTmp == null) { + flatMessageTmp = new FlatMessage(flatMessage.getId()); + partitionMessages[pkHash] = flatMessageTmp; + flatMessageTmp.setDatabase(flatMessage.getDatabase()); + flatMessageTmp.setTable(flatMessage.getTable()); + flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); + flatMessageTmp.setType(flatMessage.getType()); + flatMessageTmp.setSql(flatMessage.getSql()); + flatMessageTmp.setSqlType(flatMessage.getSqlType()); + flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); + flatMessageTmp.setEs(flatMessage.getEs()); + flatMessageTmp.setTs(flatMessage.getTs()); + } + List> data = flatMessageTmp.getData(); + if (data == null) { + data = new ArrayList<>(); + flatMessageTmp.setData(data); + } + data.add(row); + if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { + List> old = flatMessageTmp.getOld(); + if (old == null) { + old = new ArrayList<>(); + flatMessageTmp.setOld(old); + } + old.add(flatMessage.getOld().get(idx)); + } + idx++; } - old.add(flatMessage.getOld().get(idx)); } - idx++; } } } diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index a337ccc6..7e82f9c6 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -28,11 +28,11 @@ public class MQProperties { public static class CanalDestination { - private String canalDestination; - private String topic; - private Integer partition; - private Integer partitionsNum; - private Map partitionHash; + private String canalDestination; + private String topic; + private Integer partition; + private Integer partitionsNum; + private String partitionHash; public String getCanalDestination() { return canalDestination; @@ -66,11 +66,11 @@ public class MQProperties { this.partitionsNum = partitionsNum; } - public Map getPartitionHash() { + public String getPartitionHash() { return partitionHash; } - public void setPartitionHash(Map partitionHash) { + public void setPartitionHash(String partitionHash) { this.partitionHash = partitionHash; } } @@ -186,6 +186,7 @@ public class MQProperties { public void setAliyunSecretKey(String aliyunSecretKey) { this.aliyunSecretKey = aliyunSecretKey; } + public int getMaxRequestSize() { return maxRequestSize; } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index a283e23f..248e5af4 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -138,7 +138,7 @@ public class CanalMQStarter { canalDestination.setTopic(mqConfig.getTopic()); canalDestination.setPartition(mqConfig.getPartition()); canalDestination.setPartitionsNum(mqConfig.getPartitionsNum()); - canalDestination.setPartitionHash(mqConfig.getPartitionHashProperties()); + canalDestination.setPartitionHash(mqConfig.getPartitionHash()); canalServer.subscribe(clientIdentity); logger.info("## the MQ producer: {} is running now ......", destination); From 547cbf82c83eb4987e9568569788818fe1e7a806 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 01:03:18 +0800 Subject: [PATCH 07/15] =?UTF-8?q?=E5=AF=B9=E5=8E=9F=E7=94=9Fmessage?= =?UTF-8?q?=E5=8F=91=E9=80=81mq=E7=9A=84=E5=88=86=E5=8C=BA=E6=94=AF?= =?UTF-8?q?=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/protocol/FlatMessage.java | 42 +++--- .../alibaba/otter/canal/protocol/Message.java | 113 +++++++++++++++++ .../otter/canal/kafka/CanalKafkaProducer.java | 30 +++-- .../canal/rocketmq/CanalRocketMQProducer.java | 120 ++++++++++++------ 4 files changed, 240 insertions(+), 65 deletions(-) diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index 568d689a..05d8f795 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -14,24 +14,24 @@ import com.google.protobuf.ByteString; */ public class FlatMessage implements Serializable { - private static final long serialVersionUID = -3386650678735860050L; + private static final long serialVersionUID = -3386650678735860050L; - private long id; - private String database; - private String table; - private Boolean isDdl; - private String type; + private long id; + private String database; + private String table; + private Boolean isDdl; + private String type; // binlog executeTime - private Long es; + private Long es; // dml build timeStamp - private Long ts; - private String sql; - private Map sqlType; - private Map mysqlType; - private List> data; - private List> old; + private Long ts; + private String sql; + private Map sqlType; + private Map mysqlType; + private List> data; + private List> old; - private transient Message message; // 所属message + private transient CanalEntry.Entry entry; // 所属entry public FlatMessage(long id){ this.id = id; @@ -133,12 +133,12 @@ public class FlatMessage implements Serializable { this.es = es; } - public Message getMessage() { - return message; + public CanalEntry.Entry getEntry() { + return entry; } - public void setMessage(Message message) { - this.message = message; + public void setEntry(CanalEntry.Entry entry) { + this.entry = entry; } /** @@ -192,7 +192,7 @@ public class FlatMessage implements Serializable { flatMessage.setEs(entry.getHeader().getExecuteTime()); flatMessage.setTs(System.currentTimeMillis()); flatMessage.setSql(rowChange.getSql()); - flatMessage.setMessage(message); + flatMessage.setEntry(entry); if (!rowChange.getIsDdl()) { Map sqlType = new LinkedHashMap<>(); @@ -287,7 +287,7 @@ public class FlatMessage implements Serializable { if (flatMessage.getIsDdl()) { partitionMessages[0] = flatMessage; } else { - if (flatMessage.getData() != null) { + if (flatMessage.getData() != null && !flatMessage.getData().isEmpty()) { String database = flatMessage.getDatabase(); String table = flatMessage.getTable(); @@ -314,7 +314,7 @@ public class FlatMessage implements Serializable { } else { if (pk == null) { // 如果未指定主键(通配符主键)，从原生message中取主键字段 - CanalEntry.Entry entry = flatMessage.getMessage().getEntries().get(0); + CanalEntry.Entry entry = flatMessage.getEntry(); CanalEntry.RowChange rowChange; try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index bf79ee45..d8957d92 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -3,7 +3,9 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.ArrayList; import java.util.List; +import java.util.regex.Pattern; +import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; @@ -89,4 +91,115 @@ public class Message implements Serializable { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } + /** + * 将 message 分区 + * + * @param partitionsNum 分区数 + * @param pkHashConfigs 分区库表主键正则表达式 + * @return 分区message数组 + */ + @SuppressWarnings("unchecked") + public Message[] messagePartition(Integer partitionsNum, String pkHashConfigs) { + if (partitionsNum == null) { + partitionsNum = 1; + } + Message[] partitionMessages = new Message[partitionsNum]; + List[] partitionEntries = new List[partitionsNum]; + for (int i = 0; i < partitionsNum; i++) { + partitionEntries[i] = new ArrayList<>(); + } + for (Entry entry : this.getEntries()) { + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + if (rowChange.getIsDdl()) { + partitionEntries[0].add(entry); + } else { + if (rowChange.getRowDatasList() != null && !rowChange.getRowDatasList().isEmpty()) { + String database = entry.getHeader().getSchemaName(); + String table = entry.getHeader().getTableName(); + + String pk = null; + boolean isMatch = false; + + String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, ","); + for (String pkHashConfig : pkHashConfigArray) { + int i = pkHashConfig.lastIndexOf("."); + if (!pkHashConfig.endsWith(".$pk$")) { + // 如果指定了主键 + pk = pkHashConfig.substring(i + 1); + } + pkHashConfig = pkHashConfig.substring(0, i); + isMatch = Pattern.matches(pkHashConfig, database + "." + table); + if (isMatch) { + break; + } + } + + if (!isMatch) { + // 如果都没有匹配，发送到第一个分区 + partitionEntries[0].add(entry); + } else { + if (pk == null) { + // 如果未指定主键(通配符主键)，取主键字段 + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pk = column.getName(); + break; + } + } + } + } + if (pk == null) { + // 如果都没有匹配的主键，发送到第一个分区 + partitionEntries[0].add(entry); + } else { + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + boolean hasOldPk = false; + // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + if (column.getName().equalsIgnoreCase(pk)) { + int hash = column.getValue().hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + hasOldPk = true; + break; + } + } + if (!hasOldPk) { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getName().equalsIgnoreCase(pk)) { + int hash = column.getValue().hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + break; + } + } + } + } + } + } + } + } + + for (int i = 0; i < partitionsNum; i++) { + List entries = partitionEntries[i]; + if (!entries.isEmpty()) { + partitionMessages[i] = new Message(this.id, entries); + } + } + + return partitionMessages; + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 230e9607..0dd93ea5 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -80,22 +80,36 @@ public class CanalKafkaProducer implements CanalMQProducer { // producer.beginTransaction(); if (!kafkaProperties.getFlatMessage()) { try { - ProducerRecord record; + ProducerRecord record = null; if (canalDestination.getPartition() != null) { - record = new ProducerRecord(canalDestination.getTopic(), + record = new ProducerRecord<>(canalDestination.getTopic(), canalDestination.getPartition(), null, message); } else { - record = new ProducerRecord(canalDestination.getTopic(), 0, null, message); + if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { + Message[] messages = message.messagePartition(canalDestination.getPartitionsNum(), + canalDestination.getPartitionHash()); + int length = messages.length; + for (int i = 0; i < length; i++) { + Message messagePartition = messages[i]; + if (messagePartition != null) { + record = new ProducerRecord<>(canalDestination.getTopic(), i, null, messagePartition); + } + } + } else { + record = new ProducerRecord<>(canalDestination.getTopic(), 0, null, message); + } } - producer.send(record).get(); + if (record != null) { + producer.send(record).get(); - if (logger.isDebugEnabled()) { - logger.debug("Send message to kafka topic: [{}], packet: {}", - canalDestination.getTopic(), - message.toString()); + if (logger.isDebugEnabled()) { + logger.debug("Send message to kafka topic: [{}], packet: {}", + canalDestination.getTopic(), + message.toString()); + } } } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index 74e3920b..56648d4e 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -1,14 +1,7 @@ package com.alibaba.otter.canal.rocketmq; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.serializer.SerializerFeature; -import com.alibaba.otter.canal.common.CanalMessageSerializer; -import com.alibaba.otter.canal.common.MQProperties; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.server.exception.CanalServerException; -import com.alibaba.otter.canal.spi.CanalMQProducer; -import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials; -import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook; +import java.util.List; + import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; @@ -20,7 +13,15 @@ import org.apache.rocketmq.remoting.exception.RemotingException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; +import com.alibaba.otter.canal.common.CanalMessageSerializer; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.server.exception.CanalServerException; +import com.alibaba.otter.canal.spi.CanalMQProducer; +import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials; +import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook; public class CanalRocketMQProducer implements CanalMQProducer { @@ -34,8 +35,8 @@ public class CanalRocketMQProducer implements CanalMQProducer { public void init(MQProperties rocketMQProperties) { this.mqProperties = rocketMQProperties; RPCHook rpcHook = null; - if(rocketMQProperties.getAliyunAccessKey().length() > 0 - && rocketMQProperties.getAliyunSecretKey().length() > 0){ + if (rocketMQProperties.getAliyunAccessKey().length() > 0 + && rocketMQProperties.getAliyunSecretKey().length() > 0) { SessionCredentials sessionCredentials = new SessionCredentials(); sessionCredentials.setAccessKey(rocketMQProperties.getAliyunAccessKey()); sessionCredentials.setSecretKey(rocketMQProperties.getAliyunSecretKey()); @@ -59,23 +60,65 @@ public class CanalRocketMQProducer implements CanalMQProducer { Callback callback) { if (!mqProperties.getFlatMessage()) { try { - Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data, - mqProperties.isFilterTransactionEntry())); - logger.debug("send message:{} to destination:{}, partition: {}", - message, - destination.getCanalDestination(), - destination.getPartition()); - this.defaultMQProducer.send(message, new MessageQueueSelector() { - - @Override - public MessageQueue select(List mqs, Message msg, Object arg) { - int partition = 0; - if (destination.getPartition() != null) { - partition = destination.getPartition(); - } - return mqs.get(partition); + if (destination.getPartition() != null) { + Message message = new Message(destination.getTopic(), + CanalMessageSerializer.serializer(data, mqProperties.isFilterTransactionEntry())); + if (logger.isDebugEnabled()) { + logger.debug("send message:{} to destination:{}, partition: {}", + message, + destination.getCanalDestination(), + destination.getPartition()); } - }, null); + this.defaultMQProducer.send(message, new MessageQueueSelector() { + + @Override + public MessageQueue select(List mqs, Message msg, Object arg) { + int partition = 0; + if (destination.getPartition() != null) { + partition = destination.getPartition(); + } + return mqs.get(partition); + } + }, null); + } else { + if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { + com.alibaba.otter.canal.protocol.Message[] messages = data + .messagePartition(destination.getPartitionsNum(), destination.getPartitionHash()); + int length = messages.length; + for (int i = 0; i < length; i++) { + com.alibaba.otter.canal.protocol.Message dataPartition = messages[i]; + if (dataPartition != null) { + if (logger.isDebugEnabled()) { + logger.debug("flatMessagePart: {}, partition: {}", + JSON.toJSONString(dataPartition, SerializerFeature.WriteMapNullValue), + i); + } + final int index = i; + try { + Message message = new Message(destination.getTopic(), + CanalMessageSerializer.serializer(dataPartition, + mqProperties.isFilterTransactionEntry())); + this.defaultMQProducer.send(message, new MessageQueueSelector() { + + @Override + public MessageQueue select(List mqs, Message msg, Object arg) { + if (index > mqs.size()) { + throw new CanalServerException("partition number is error,config num:" + + destination.getPartitionsNum() + + ", mq num: " + mqs.size()); + } + return mqs.get(index); + } + }, null); + } catch (Exception e) { + logger.error("send flat message to hashed partition error", e); + callback.rollback(); + return; + } + } + } + } + } } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { logger.error("Send message error!", e); callback.rollback(); @@ -87,10 +130,12 @@ public class CanalRocketMQProducer implements CanalMQProducer { for (FlatMessage flatMessage : flatMessages) { if (destination.getPartition() != null) { try { - logger.info("send flat message: {} to topic: {} fixed partition: {}", - JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue), - destination.getTopic(), - destination.getPartition()); + if (logger.isDebugEnabled()) { + logger.debug("send message: {} to topic: {} fixed partition: {}", + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue), + destination.getTopic(), + destination.getPartition()); + } Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @@ -114,13 +159,16 @@ public class CanalRocketMQProducer implements CanalMQProducer { for (int i = 0; i < length; i++) { FlatMessage flatMessagePart = partitionFlatMessage[i]; if (flatMessagePart != null) { - logger.debug("flatMessagePart: {}, partition: {}", - JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue), - i); + if (logger.isDebugEnabled()) { + logger.debug("flatMessagePart: {}, partition: {}", + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue), + i); + } final int index = i; try { Message message = new Message(destination.getTopic(), - JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue).getBytes()); + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue) + .getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override From 5650b56bedc4eeb919a1f1389b5d67adf57be6c3 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 01:10:04 +0800 Subject: [PATCH 08/15] =?UTF-8?q?=E6=89=93=E5=8D=B0=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/launcher/loader/AbstractCanalAdapterWorker.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index db87bf01..ec650231 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -70,6 +70,7 @@ public abstract class AbstractCanalAdapterWorker { }); return true; } catch (Exception e) { + logger.error(e.getMessage(), e); return false; } })); @@ -108,6 +109,7 @@ public abstract class AbstractCanalAdapterWorker { }); return true; } catch (Exception e) { + logger.error(e.getMessage(), e); return false; } })); From 99a1123e010eea0b087c11421932c820c34a6681 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 10 Dec 2018 09:52:48 +0800 Subject: [PATCH 09/15] =?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 --- .../resources/example/instance.properties | 6 ++-- .../alibaba/otter/canal/protocol/Message.java | 29 +++++++++++++++---- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 3a9ac883..8d884abd 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -1,5 +1,5 @@ ################################################# -## mysql serverId , v1.0.26+ will autoGen +## mysql serverId , v1.0.26+ will autoGen # canal.instance.mysql.slaveId=0 # enable gtid use true/false @@ -48,5 +48,5 @@ canal.mq.topic=example canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=3 -#canal.mq.partitionHash=mytest.person:id,mytest.role:id -################################################# \ No newline at end of file +#canal.mq.partitionHash=.*\\..*.$pk$ +################################################# diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index d8957d92..a01a8006 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -5,6 +5,7 @@ import java.util.ArrayList; import java.util.List; import java.util.regex.Pattern; +import com.google.protobuf.InvalidProtocolBufferException; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; @@ -93,7 +94,7 @@ public class Message implements Serializable { /** * 将 message 分区 - * + * * @param partitionsNum 分区数 * @param pkHashConfigs 分区库表主键正则表达式 * @return 分区message数组 @@ -108,7 +109,25 @@ public class Message implements Serializable { for (int i = 0; i < partitionsNum; i++) { partitionEntries[i] = new ArrayList<>(); } - for (Entry entry : this.getEntries()) { + + List entries; + if (this.isRaw()) { + List rawEntries = this.getRawEntries(); + entries = new ArrayList<>(rawEntries.size()); + for (ByteString byteString : rawEntries) { + Entry entry; + try { + entry = Entry.parseFrom(byteString); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + entries.add(entry); + } + } else { + entries = this.getEntries(); + } + + for (Entry entry : entries) { CanalEntry.RowChange rowChange; try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); @@ -194,9 +213,9 @@ public class Message implements Serializable { } for (int i = 0; i < partitionsNum; i++) { - List entries = partitionEntries[i]; - if (!entries.isEmpty()) { - partitionMessages[i] = new Message(this.id, entries); + List entriesTmp = partitionEntries[i]; + if (!entriesTmp.isEmpty()) { + partitionMessages[i] = new Message(this.id, entriesTmp); } } From 7c63e81e01a2acb7385c0511232d39df360a4001 Mon Sep 17 00:00:00 2001 From: agapple Date: Mon, 10 Dec 2018 14:42:28 +0800 Subject: [PATCH 10/15] fixed issue #1256, rds alibaba_rds_row_id --- .../inbound/mysql/dbsync/LogEventConvert.java | 36 ++++++++++++++----- 1 file changed, 27 insertions(+), 9 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 3f811971..ded1343d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -621,9 +621,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar + "," + columnInfo.length + " vs " + tableMeta.getFields().size()); } } - } else { - logger.warn("[" + event.getTable().getDbName() + "." + event.getTable().getTableName() - + "] is no primary key , skip alibaba_rds_row_id column"); + // } else { + // logger.warn("[" + event.getTable().getDbName() + "." + + // event.getTable().getTableName() + // + "] is no primary key , skip alibaba_rds_row_id column"); } } @@ -634,18 +635,35 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar continue; } + if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { + // 不解析最后一列 + String rdsRowIdColumnName = "#alibaba_rds_row_id#"; + buffer.nextValue(rdsRowIdColumnName, i, info.type, info.meta, false); + Column.Builder columnBuilder = Column.newBuilder(); + columnBuilder.setName(rdsRowIdColumnName); + columnBuilder.setIsKey(true); + columnBuilder.setMysqlType("bigint"); + columnBuilder.setIndex(i); + columnBuilder.setIsNull(false); + Serializable value = buffer.getValue(); + columnBuilder.setValue(value.toString()); + columnBuilder.setSqlType(Types.BIGINT); + columnBuilder.setUpdated(false); + + if (isAfter) { + rowDataBuilder.addAfterColumns(columnBuilder.build()); + } else { + rowDataBuilder.addBeforeColumns(columnBuilder.build()); + } + continue; + } + FieldMeta fieldMeta = null; if (tableMeta != null && !tableError) { // 处理file meta fieldMeta = tableMeta.getFields().get(i); } - if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { - // 不解析最后一列 - buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, false); - continue; - } - if (fieldMeta != null && existOptionalMetaData && tableMetaCache.isOnTSDB()) { // check column info boolean check = StringUtils.equalsIgnoreCase(fieldMeta.getColumnName(), info.name); From 706ee88914c738073b0ae34180f10ce90fbb845b Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 10 Dec 2018 21:28:09 +0800 Subject: [PATCH 11/15] =?UTF-8?q?pattern=E8=A1=A8=E8=BE=BE=E5=BC=8F?= =?UTF-8?q?=E8=B0=83=E6=95=B4=20pk=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- protocol/pom.xml | 8 ++ .../otter/canal/protocol/FlatMessage.java | 78 ++++++----- .../alibaba/otter/canal/protocol/Message.java | 54 +++++--- .../protocol/aviater/AviaterRegexFilter.java | 124 ++++++++++++++++++ .../canal/protocol/aviater/PatternUtils.java | 38 ++++++ .../canal/protocol/aviater/RegexFunction.java | 31 +++++ 6 files changed, 283 insertions(+), 50 deletions(-) create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java diff --git a/protocol/pom.xml b/protocol/pom.xml index 1b776b7e..0f7904d6 100644 --- a/protocol/pom.xml +++ b/protocol/pom.xml @@ -26,5 +26,13 @@ commons-lang commons-lang + + oro + oro + + + com.googlecode.aviator + aviator + diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index 05d8f795..263b0d6e 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -2,8 +2,10 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.*; -import java.util.regex.Pattern; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; import com.google.protobuf.ByteString; @@ -14,24 +16,27 @@ import com.google.protobuf.ByteString; */ public class FlatMessage implements Serializable { - private static final long serialVersionUID = -3386650678735860050L; + private static final long serialVersionUID = -3386650678735860050L; - private long id; - private String database; - private String table; - private Boolean isDdl; - private String type; + private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); + private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); + + private long id; + private String database; + private String table; + private Boolean isDdl; + private String type; // binlog executeTime - private Long es; + private Long es; // dml build timeStamp - private Long ts; - private String sql; - private Map sqlType; - private Map mysqlType; - private List> data; - private List> old; + private Long ts; + private String sql; + private Map sqlType; + private Map mysqlType; + private List> data; + private List> old; - private transient CanalEntry.Entry entry; // 所属entry + private transient CanalEntry.Entry entry; // 所属entry public FlatMessage(long id){ this.id = id; @@ -143,7 +148,7 @@ public class FlatMessage implements Serializable { /** * 将Message转换为FlatMessage - * + * * @param message 原生message * @return FlatMessage列表 */ @@ -272,7 +277,7 @@ public class FlatMessage implements Serializable { /** * 将FlatMessage按指定的字段值hash拆分 - * + * * @param flatMessage flatMessage * @param partitionsNum 分区数量 * @param pkHashConfigs hash映射 @@ -302,7 +307,14 @@ public class FlatMessage implements Serializable { pk = pkHashConfig.substring(i + 1); } pkHashConfig = pkHashConfig.substring(0, i); - isMatch = Pattern.matches(pkHashConfig, database + "." + table); + + AviaterRegexFilter aviaterRegexFilter = regexFilters.get(pkHashConfig); + if (aviaterRegexFilter == null) { + aviaterRegexFilter = new AviaterRegexFilter(pkHashConfig); + regexFilters.putIfAbsent(pkHashConfig, aviaterRegexFilter); + } + + isMatch = aviaterRegexFilter.filter(database + "." + table); if (isMatch) { break; } @@ -313,19 +325,23 @@ public class FlatMessage implements Serializable { partitionMessages[0] = flatMessage; } else { if (pk == null) { - // 如果未指定主键(通配符主键)，从原生message中取主键字段 - CanalEntry.Entry entry = flatMessage.getEntry(); - CanalEntry.RowChange rowChange; - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException(e.getMessage(), e); - } - CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getIsKey()) { - pk = column.getName(); - break; + pk = schemaTabPk.get(database + "." + table); + if (pk == null) { + // 如果未指定主键(通配符主键)，从原生message中取主键字段 + CanalEntry.Entry entry = flatMessage.getEntry(); + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pk = column.getName(); + schemaTabPk.putIfAbsent(database + "." + table, pk); + break; + } } } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index a01a8006..00837fec 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -3,15 +3,17 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.ArrayList; import java.util.List; -import java.util.regex.Pattern; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; -import com.google.protobuf.InvalidProtocolBufferException; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; /** * @author zebin.xuzb @ 2012-6-19 @@ -19,14 +21,17 @@ import com.google.protobuf.ByteString; */ public class Message implements Serializable { - private static final long serialVersionUID = 1234034768477580009L; + private static final long serialVersionUID = 1234034768477580009L; - private long id; - private List entries = new ArrayList(); + private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); + private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); + + private long id; + private List entries = new ArrayList(); // row data for performance, see: // https://github.com/alibaba/canal/issues/726 - private boolean raw = true; - private List rawEntries = new ArrayList(); + private boolean raw = true; + private List rawEntries = new ArrayList(); public Message(long id, List entries){ this.id = id; @@ -152,7 +157,14 @@ public class Message implements Serializable { pk = pkHashConfig.substring(i + 1); } pkHashConfig = pkHashConfig.substring(0, i); - isMatch = Pattern.matches(pkHashConfig, database + "." + table); + + AviaterRegexFilter aviaterRegexFilter = regexFilters.get(pkHashConfig); + if (aviaterRegexFilter == null) { + aviaterRegexFilter = new AviaterRegexFilter(pkHashConfig); + regexFilters.putIfAbsent(pkHashConfig, aviaterRegexFilter); + } + + isMatch = aviaterRegexFilter.filter(database + "." + table); if (isMatch) { break; } @@ -163,17 +175,21 @@ public class Message implements Serializable { partitionEntries[0].add(entry); } else { if (pk == null) { - // 如果未指定主键(通配符主键)，取主键字段 - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException(e.getMessage(), e); - } - CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getIsKey()) { - pk = column.getName(); - break; + pk = schemaTabPk.get(database + "." + table); + if (pk == null) { + // 如果未指定主键(通配符主键)，取主键字段 + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pk = column.getName(); + schemaTabPk.putIfAbsent(database + "." + table, pk); + break; + } } } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java new file mode 100644 index 00000000..db7750ba --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java @@ -0,0 +1,124 @@ +package com.alibaba.otter.canal.protocol.aviater; + +import java.util.*; + +import org.apache.commons.lang.StringUtils; + +import com.googlecode.aviator.AviatorEvaluator; +import com.googlecode.aviator.Expression; + +/** + * 基于aviater进行tableName正则匹配的过滤算法 + * + * @author jianghang 2012-7-20 下午06:01:34 + */ +public class AviaterRegexFilter { + + private static final String SPLIT = ","; + private static final String PATTERN_SPLIT = "|"; + private static final String FILTER_EXPRESSION = "regex(pattern,target)"; + private static final RegexFunction regexFunction = new RegexFunction(); + private final Expression exp = AviatorEvaluator.compile(FILTER_EXPRESSION, true); + static { + AviatorEvaluator.addFunction(regexFunction); + } + + private static final Comparator COMPARATOR = new StringComparator(); + + final private String pattern; + final private boolean defaultEmptyValue; + + public AviaterRegexFilter(String pattern){ + this(pattern, true); + } + + public AviaterRegexFilter(String pattern, boolean defaultEmptyValue){ + this.defaultEmptyValue = defaultEmptyValue; + List list = null; + if (StringUtils.isEmpty(pattern)) { + list = new ArrayList(); + } else { + String[] ss = StringUtils.split(pattern, SPLIT); + list = Arrays.asList(ss); + } + + // 对pattern按照从长到短的排序 + // 因为 foo|foot 匹配 foot 会出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot + // 的长度不一样 + Collections.sort(list, COMPARATOR); + // 对pattern进行头尾完全匹配 + list = completionPattern(list); + this.pattern = StringUtils.join(list, PATTERN_SPLIT); + } + + public boolean filter(String filtered) { + if (StringUtils.isEmpty(pattern)) { + return defaultEmptyValue; + } + + if (StringUtils.isEmpty(filtered)) { + return defaultEmptyValue; + } + + Map env = new HashMap(); + env.put("pattern", pattern); + env.put("target", filtered.toLowerCase()); + return (Boolean) exp.execute(env); + } + + /** + * 修复正则表达式匹配的问题，因为使用了 oro 的 matches，会出现： + * + * 
+     * foo|foot 匹配 foot 出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot 的长度不一样
+     *
+ * + * 因此此类对正则表达式进行了从长到短的排序 + * + * @author zebin.xuzb 2012-10-22 下午2:02:26 + * @version 1.0.0 + */ + private static class StringComparator implements Comparator { + + @Override + public int compare(String str1, String str2) { + if (str1.length() > str2.length()) { + return -1; + } else if (str1.length() < str2.length()) { + return 1; + } else { + return 0; + } + } + } + + /** + * 修复正则表达式匹配的问题，即使按照长度递减排序，还是会出现以下问题： + * + * 
+     * foooo|f.*t 匹配 fooooot 出错，原因是 fooooot 匹配了 foooo 之后，会将 fooo 和数据进行匹配，但是 foooo 的长度和 fooooot 的长度不一样
+     *
+ * + * 因此此类对正则表达式进行头尾完全匹配 + * + * @author simon + * @version 1.0.0 + */ + + private List completionPattern(List patterns) { + List result = new ArrayList(); + for (String pattern : patterns) { + StringBuffer stringBuffer = new StringBuffer(); + stringBuffer.append("^"); + stringBuffer.append(pattern); + stringBuffer.append("$"); + result.add(stringBuffer.toString()); + } + return result; + } + + @Override + public String toString() { + return pattern; + } +} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java new file mode 100644 index 00000000..e04f0c4b --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java @@ -0,0 +1,38 @@ +package com.alibaba.otter.canal.protocol.aviater; + +import java.util.Map; + +import org.apache.oro.text.regex.MalformedPatternException; +import org.apache.oro.text.regex.Pattern; +import org.apache.oro.text.regex.PatternCompiler; +import org.apache.oro.text.regex.Perl5Compiler; + +import com.google.common.base.Function; +import com.google.common.collect.MapMaker; +import com.google.common.collect.MigrateMap; + +public class PatternUtils { + @SuppressWarnings("deprecation") + private static Map patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(), + new Function() { + + public Pattern apply(String pattern) { + try { + PatternCompiler pc = new Perl5Compiler(); + return pc.compile(pattern, + Perl5Compiler.CASE_INSENSITIVE_MASK | Perl5Compiler.READ_ONLY_MASK + | Perl5Compiler.SINGLELINE_MASK); + } catch (MalformedPatternException e) { + throw new RuntimeException(e); + } + } + }); + + public static Pattern getPattern(String pattern) { + return patterns.get(pattern); + } + + public static void clear() { + patterns.clear(); + } +} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java new file mode 100644 index 00000000..be084c25 --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java @@ -0,0 +1,31 @@ +package com.alibaba.otter.canal.protocol.aviater; + +import java.util.Map; + +import org.apache.oro.text.regex.Perl5Matcher; + +import com.googlecode.aviator.runtime.function.AbstractFunction; +import com.googlecode.aviator.runtime.function.FunctionUtils; +import com.googlecode.aviator.runtime.type.AviatorBoolean; +import com.googlecode.aviator.runtime.type.AviatorObject; + +/** + * 提供aviator regex的代码扩展 + * + * @author jianghang 2012-7-23 上午10:29:23 + */ +public class RegexFunction extends AbstractFunction { + + public AviatorObject call(Map env, AviatorObject arg1, AviatorObject arg2) { + String pattern = FunctionUtils.getStringValue(arg1, env); + String text = FunctionUtils.getStringValue(arg2, env); + Perl5Matcher matcher = new Perl5Matcher(); + boolean isMatch = matcher.matches(text, PatternUtils.getPattern(pattern)); + return AviatorBoolean.valueOf(isMatch); + } + + public String getName() { + return "regex"; + } + +} From eac3f5ce5fa2152136e525e1909c91d7373381f8 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 23:54:07 +0800 Subject: [PATCH 12/15] =?UTF-8?q?=E4=BF=AE=E6=94=B9convert=E6=97=B6?= =?UTF-8?q?=E8=AE=BE=E7=BD=AEpk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/protocol/FlatMessage.java | 35 ++++----- .../alibaba/otter/canal/protocol/Message.java | 77 +++++++++---------- 2 files changed, 49 insertions(+), 63 deletions(-) diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index 263b0d6e..53924d4a 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -5,9 +5,9 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import com.google.protobuf.ByteString; /** @@ -18,12 +18,12 @@ public class FlatMessage implements Serializable { private static final long serialVersionUID = -3386650678735860050L; - private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); private long id; private String database; private String table; + private String pk; private Boolean isDdl; private String type; // binlog executeTime @@ -66,6 +66,14 @@ public class FlatMessage implements Serializable { this.table = table; } + public String getPk() { + return pk; + } + + public void setPk(String pk) { + this.pk = pk; + } + public Boolean getIsDdl() { return isDdl; } @@ -222,6 +230,9 @@ public class FlatMessage implements Serializable { } for (CanalEntry.Column column : columns) { + if (flatMessage.getPk() == null && column.getIsKey()) { + flatMessage.setPk(column.getName()); + } sqlType.put(column.getName(), column.getSqlType()); mysqlType.put(column.getName(), column.getMysqlType()); if (column.getIsNull()) { @@ -325,25 +336,7 @@ public class FlatMessage implements Serializable { partitionMessages[0] = flatMessage; } else { if (pk == null) { - pk = schemaTabPk.get(database + "." + table); - if (pk == null) { - // 如果未指定主键(通配符主键)，从原生message中取主键字段 - CanalEntry.Entry entry = flatMessage.getEntry(); - CanalEntry.RowChange rowChange; - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException(e.getMessage(), e); - } - CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getIsKey()) { - pk = column.getName(); - schemaTabPk.putIfAbsent(database + "." + table, pk); - break; - } - } - } + pk = flatMessage.getPk(); } if (pk == null || !flatMessage.getData().get(0).containsKey(pk)) { // 如果都没有匹配的主键，发送到第一个分区 diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 00837fec..6fc42a41 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -4,14 +4,13 @@ import java.io.Serializable; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; @@ -23,7 +22,6 @@ public class Message implements Serializable { private static final long serialVersionUID = 1234034768477580009L; - private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); private long id; @@ -173,55 +171,50 @@ public class Message implements Serializable { if (!isMatch) { // 如果都没有匹配，发送到第一个分区 partitionEntries[0].add(entry); - } else { - if (pk == null) { - pk = schemaTabPk.get(database + "." + table); - if (pk == null) { - // 如果未指定主键(通配符主键)，取主键字段 - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException(e.getMessage(), e); + } + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + boolean hasOldPk = false; + // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + CanalEntry.Column pkColumn = null; + if (pk != null) { + if (column.getName().equalsIgnoreCase(pk)) { + pkColumn = column; } - CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getIsKey()) { - pk = column.getName(); - schemaTabPk.putIfAbsent(database + "." + table, pk); - break; - } + } else { + if (column.getIsKey()) { + pkColumn = column; } } + if (pkColumn != null) { + int hash = pkColumn.getValue().hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + hasOldPk = true; + break; + } } - } - if (pk == null) { - // 如果都没有匹配的主键，发送到第一个分区 - partitionEntries[0].add(entry); - } else { - for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - boolean hasOldPk = false; - // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 - for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { - if (column.getName().equalsIgnoreCase(pk)) { - int hash = column.getValue().hashCode(); + if (!hasOldPk) { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + CanalEntry.Column pkColumn = null; + if (pk != null) { + if (column.getName().equalsIgnoreCase(pk)) { + pkColumn = column; + } + } else { + if (column.getIsKey()) { + pkColumn = column; + } + } + if (pkColumn != null) { + int hash = pkColumn.getValue().hashCode(); int pkHash = Math.abs(hash) % partitionsNum; pkHash = Math.abs(pkHash); partitionEntries[pkHash].add(entry); - hasOldPk = true; break; } } - if (!hasOldPk) { - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getName().equalsIgnoreCase(pk)) { - int hash = column.getValue().hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); - break; - } - } - } } } } From aeecf87490eee5dc297dfea7d3b5f25c660d8a80 Mon Sep 17 00:00:00 2001 From: machey Date: Tue, 11 Dec 2018 00:09:12 +0800 Subject: [PATCH 13/15] =?UTF-8?q?=E5=8E=BB=E6=8E=89=E5=AF=B9before?= =?UTF-8?q?=E7=9A=84=E5=88=A4=E6=96=AD=EF=BC=8C=E5=88=86=E5=8C=BA=E5=8F=91?= =?UTF-8?q?=E9=80=81=E7=9A=84=E8=A1=A8pk=E4=B8=8D=E8=83=BD=E4=BF=AE?= =?UTF-8?q?=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/protocol/FlatMessage.java | 12 +--- .../alibaba/otter/canal/protocol/Message.java | 58 +++++++------------ 2 files changed, 21 insertions(+), 49 deletions(-) diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index 53924d4a..f46c26a4 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -344,17 +344,7 @@ public class FlatMessage implements Serializable { } else { int idx = 0; for (Map row : flatMessage.getData()) { - Map o = null; - if (flatMessage.getOld() != null) { - o = flatMessage.getOld().get(idx); - } - String value; - // 如果old中有pk值说明主键有修改, 以旧的主键值hash为准 - if (o != null && o.containsKey(pk)) { - value = o.get(pk); - } else { - value = row.get(pk); - } + String value = row.get(pk); if (value == null) { value = ""; } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 6fc42a41..1013bfe6 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -173,48 +173,30 @@ public class Message implements Serializable { partitionEntries[0].add(entry); } for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - boolean hasOldPk = false; - // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 - for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { - CanalEntry.Column pkColumn = null; - if (pk != null) { - if (column.getName().equalsIgnoreCase(pk)) { - pkColumn = column; - } - } else { - if (column.getIsKey()) { - pkColumn = column; - } - } - if (pkColumn != null) { - int hash = pkColumn.getValue().hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); - hasOldPk = true; - break; - } - } - if (!hasOldPk) { + String pkValue = null; + if (pk != null) { for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - CanalEntry.Column pkColumn = null; - if (pk != null) { - if (column.getName().equalsIgnoreCase(pk)) { - pkColumn = column; - } - } else { - if (column.getIsKey()) { - pkColumn = column; - } - } - if (pkColumn != null) { - int hash = pkColumn.getValue().hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); + if (column.getName().equalsIgnoreCase(pk)) { + pkValue = column.getValue(); break; } } + } else { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pkValue = column.getValue(); + break; + } + } + } + if (pkValue != null) { + int hash = pkValue.hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + } else { + // 没有找到主键 + partitionEntries[0].add(entry); } } } From 277a273ebc71ada14372698fcdbd83ff120d45bd Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 11 Dec 2018 13:40:32 +0800 Subject: [PATCH 14/15] fixed pull #1266 , optimizer code --- .../resources/example/instance.properties | 2 +- .../otter/canal/filter/PatternUtils.java | 9 +- .../filter/aviater/AviaterRegexFilter.java | 1 + .../canal/filter/aviater/RegexFunction.java | 2 +- .../otter/canal/protocol/FlatMessage.java | 302 ++----------- .../alibaba/otter/canal/protocol/Message.java | 134 +----- .../protocol/aviater/AviaterRegexFilter.java | 124 ------ .../canal/protocol/aviater/PatternUtils.java | 38 -- .../canal/protocol/aviater/RegexFunction.java | 31 -- .../otter/canal/common/MQMessageUtils.java | 405 ++++++++++++++++++ .../otter/canal/common/MQProperties.java | 1 - .../otter/canal/kafka/CanalKafkaProducer.java | 17 +- .../canal/rocketmq/CanalRocketMQProducer.java | 27 +- 13 files changed, 464 insertions(+), 629 deletions(-) delete mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java delete mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java delete mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java create mode 100644 server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 8d884abd..2a1d8aac 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -48,5 +48,5 @@ canal.mq.topic=example canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=3 -#canal.mq.partitionHash=.*\\..*.$pk$ +#canal.mq.partitionHash=test.table#id^name,.*\\..* ################################################# diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java b/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java index d7f27503..52b192d0 100644 --- a/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java +++ b/filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java @@ -7,17 +7,10 @@ import org.apache.oro.text.regex.Pattern; import org.apache.oro.text.regex.PatternCompiler; import org.apache.oro.text.regex.Perl5Compiler; -import com.alibaba.otter.canal.filter.exception.CanalFilterException; import com.google.common.base.Function; import com.google.common.collect.MapMaker; import com.google.common.collect.MigrateMap; -/** - * 提供{@linkplain Pattern}的lazy get处理 - * - * @author jianghang 2013-1-22 下午09:36:44 - * @version 1.0.0 - */ public class PatternUtils { @SuppressWarnings("deprecation") @@ -32,7 +25,7 @@ public class PatternUtils { | Perl5Compiler.READ_ONLY_MASK | Perl5Compiler.SINGLELINE_MASK); } catch (MalformedPatternException e) { - throw new CanalFilterException(e); + throw new RuntimeException(e); } } }); diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java index 026fe006..4eb0dffa 100644 --- a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java +++ b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java @@ -129,4 +129,5 @@ public class AviaterRegexFilter implements CanalEventFilter { public String toString() { return pattern; } + } diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/RegexFunction.java b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/RegexFunction.java index ec389416..72f2136d 100644 --- a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/RegexFunction.java +++ b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/RegexFunction.java @@ -12,7 +12,7 @@ import com.googlecode.aviator.runtime.type.AviatorObject; /** * 提供aviator regex的代码扩展 - * + * * @author jianghang 2012-7-23 上午10:29:23 */ public class RegexFunction extends AbstractFunction { diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index f46c26a4..50e1cb3f 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -1,14 +1,10 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; +import java.util.List; +import java.util.Map; -import org.apache.commons.lang.StringUtils; - -import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; -import com.google.protobuf.ByteString; +import com.google.common.collect.Lists; /** * @author machengyuan 2018-9-13 下午10:31:14 @@ -16,27 +12,22 @@ import com.google.protobuf.ByteString; */ public class FlatMessage implements Serializable { - private static final long serialVersionUID = -3386650678735860050L; - - private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); - - private long id; - private String database; - private String table; - private String pk; - private Boolean isDdl; - private String type; + private static final long serialVersionUID = -3386650678735860050L; + private long id; + private String database; + private String table; + private List pkNames; + private Boolean isDdl; + private String type; // binlog executeTime - private Long es; + private Long es; // dml build timeStamp - private Long ts; - private String sql; - private Map sqlType; - private Map mysqlType; - private List> data; - private List> old; - - private transient CanalEntry.Entry entry; // 所属entry + private Long ts; + private String sql; + private Map sqlType; + private Map mysqlType; + private List> data; + private List> old; public FlatMessage(long id){ this.id = id; @@ -66,12 +57,19 @@ public class FlatMessage implements Serializable { this.table = table; } - public String getPk() { - return pk; + public List getPkNames() { + return pkNames; } - public void setPk(String pk) { - this.pk = pk; + public void addPkName(String pkName) { + if (this.pkNames == null) { + this.pkNames = Lists.newArrayList(); + } + this.pkNames.add(pkName); + } + + public void setPkNames(List pkNames) { + this.pkNames = pkNames; } public Boolean getIsDdl() { @@ -146,250 +144,6 @@ public class FlatMessage implements Serializable { this.es = es; } - public CanalEntry.Entry getEntry() { - return entry; - } - - public void setEntry(CanalEntry.Entry entry) { - this.entry = entry; - } - - /** - * 将Message转换为FlatMessage - * - * @param message 原生message - * @return FlatMessage列表 - */ - public static List messageConverter(Message message) { - try { - if (message == null) { - return null; - } - - List flatMessages = new ArrayList<>(); - List entrys = null; - if (message.isRaw()) { - List rawEntries = message.getRawEntries(); - entrys = new ArrayList(rawEntries.size()); - for (ByteString byteString : rawEntries) { - CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); - entrys.add(entry); - } - } else { - entrys = message.getEntries(); - } - - for (CanalEntry.Entry entry : entrys) { - if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN - || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { - continue; - } - - CanalEntry.RowChange rowChange; - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException( - "ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), - e); - } - - CanalEntry.EventType eventType = rowChange.getEventType(); - - FlatMessage flatMessage = new FlatMessage(message.getId()); - flatMessages.add(flatMessage); - flatMessage.setDatabase(entry.getHeader().getSchemaName()); - flatMessage.setTable(entry.getHeader().getTableName()); - flatMessage.setIsDdl(rowChange.getIsDdl()); - flatMessage.setType(eventType.toString()); - flatMessage.setEs(entry.getHeader().getExecuteTime()); - flatMessage.setTs(System.currentTimeMillis()); - flatMessage.setSql(rowChange.getSql()); - flatMessage.setEntry(entry); - - if (!rowChange.getIsDdl()) { - Map sqlType = new LinkedHashMap<>(); - Map mysqlType = new LinkedHashMap<>(); - List> data = new ArrayList<>(); - List> old = new ArrayList<>(); - - Set updateSet = new HashSet<>(); - for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE - && eventType != CanalEntry.EventType.DELETE) { - continue; - } - - Map row = new LinkedHashMap<>(); - List columns; - - if (eventType == CanalEntry.EventType.DELETE) { - columns = rowData.getBeforeColumnsList(); - } else { - columns = rowData.getAfterColumnsList(); - } - - for (CanalEntry.Column column : columns) { - if (flatMessage.getPk() == null && column.getIsKey()) { - flatMessage.setPk(column.getName()); - } - sqlType.put(column.getName(), column.getSqlType()); - mysqlType.put(column.getName(), column.getMysqlType()); - if (column.getIsNull()) { - row.put(column.getName(), null); - } else { - row.put(column.getName(), column.getValue()); - } - // 获取update为true的字段 - if (column.getUpdated()) { - updateSet.add(column.getName()); - } - } - if (!row.isEmpty()) { - data.add(row); - } - - if (eventType == CanalEntry.EventType.UPDATE) { - Map rowOld = new LinkedHashMap<>(); - for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { - if (updateSet.contains(column.getName())) { - if (column.getIsNull()) { - rowOld.put(column.getName(), null); - } else { - rowOld.put(column.getName(), column.getValue()); - } - } - } - // update操作将记录修改前的值 - if (!rowOld.isEmpty()) { - old.add(rowOld); - } - } - } - if (!sqlType.isEmpty()) { - flatMessage.setSqlType(sqlType); - } - if (!mysqlType.isEmpty()) { - flatMessage.setMysqlType(mysqlType); - } - if (!data.isEmpty()) { - flatMessage.setData(data); - } - if (!old.isEmpty()) { - flatMessage.setOld(old); - } - } - } - return flatMessages; - } catch (Exception e) { - throw new RuntimeException(e); - } - } - - /** - * 将FlatMessage按指定的字段值hash拆分 - * - * @param flatMessage flatMessage - * @param partitionsNum 分区数量 - * @param pkHashConfigs hash映射 - * @return 拆分后的flatMessage数组 - */ - public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs) { - if (partitionsNum == null) { - partitionsNum = 1; - } - FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; - - if (flatMessage.getIsDdl()) { - partitionMessages[0] = flatMessage; - } else { - if (flatMessage.getData() != null && !flatMessage.getData().isEmpty()) { - String database = flatMessage.getDatabase(); - String table = flatMessage.getTable(); - - String pk = null; - boolean isMatch = false; - - String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, ","); - for (String pkHashConfig : pkHashConfigArray) { - int i = pkHashConfig.lastIndexOf("."); - if (!pkHashConfig.endsWith(".$pk$")) { - // 如果指定了主键 - pk = pkHashConfig.substring(i + 1); - } - pkHashConfig = pkHashConfig.substring(0, i); - - AviaterRegexFilter aviaterRegexFilter = regexFilters.get(pkHashConfig); - if (aviaterRegexFilter == null) { - aviaterRegexFilter = new AviaterRegexFilter(pkHashConfig); - regexFilters.putIfAbsent(pkHashConfig, aviaterRegexFilter); - } - - isMatch = aviaterRegexFilter.filter(database + "." + table); - if (isMatch) { - break; - } - } - - if (!isMatch) { - // 如果都没有匹配，发送到第一个分区 - partitionMessages[0] = flatMessage; - } else { - if (pk == null) { - pk = flatMessage.getPk(); - } - if (pk == null || !flatMessage.getData().get(0).containsKey(pk)) { - // 如果都没有匹配的主键，发送到第一个分区 - partitionMessages[0] = flatMessage; - } else { - int idx = 0; - for (Map row : flatMessage.getData()) { - String value = row.get(pk); - if (value == null) { - value = ""; - } - int hash = value.hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 - pkHash = Math.abs(pkHash); - - FlatMessage flatMessageTmp = partitionMessages[pkHash]; - if (flatMessageTmp == null) { - flatMessageTmp = new FlatMessage(flatMessage.getId()); - partitionMessages[pkHash] = flatMessageTmp; - flatMessageTmp.setDatabase(flatMessage.getDatabase()); - flatMessageTmp.setTable(flatMessage.getTable()); - flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); - flatMessageTmp.setType(flatMessage.getType()); - flatMessageTmp.setSql(flatMessage.getSql()); - flatMessageTmp.setSqlType(flatMessage.getSqlType()); - flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); - flatMessageTmp.setEs(flatMessage.getEs()); - flatMessageTmp.setTs(flatMessage.getTs()); - } - List> data = flatMessageTmp.getData(); - if (data == null) { - data = new ArrayList<>(); - flatMessageTmp.setData(data); - } - data.add(row); - if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { - List> old = flatMessageTmp.getOld(); - if (old == null) { - old = new ArrayList<>(); - flatMessageTmp.setOld(old); - } - old.add(flatMessage.getOld().get(idx)); - } - idx++; - } - } - } - } - } - return partitionMessages; - } - @Override public String toString() { return "FlatMessage [id=" + id + ", database=" + database + ", table=" + table + ", isDdl=" + isDdl + ", type=" diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 1013bfe6..3fa691ad 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -3,16 +3,12 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.ConcurrentHashMap; -import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; -import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import com.google.protobuf.ByteString; -import com.google.protobuf.InvalidProtocolBufferException; /** * @author zebin.xuzb @ 2012-6-19 @@ -20,16 +16,13 @@ import com.google.protobuf.InvalidProtocolBufferException; */ public class Message implements Serializable { - private static final long serialVersionUID = 1234034768477580009L; - - private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); - - private long id; - private List entries = new ArrayList(); + private static final long serialVersionUID = 1234034768477580009L; + private long id; + private List entries = new ArrayList(); // row data for performance, see: // https://github.com/alibaba/canal/issues/726 - private boolean raw = true; - private List rawEntries = new ArrayList(); + private boolean raw = true; + private List rawEntries = new ArrayList(); public Message(long id, List entries){ this.id = id; @@ -95,121 +88,4 @@ public class Message implements Serializable { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } - /** - * 将 message 分区 - * - * @param partitionsNum 分区数 - * @param pkHashConfigs 分区库表主键正则表达式 - * @return 分区message数组 - */ - @SuppressWarnings("unchecked") - public Message[] messagePartition(Integer partitionsNum, String pkHashConfigs) { - if (partitionsNum == null) { - partitionsNum = 1; - } - Message[] partitionMessages = new Message[partitionsNum]; - List[] partitionEntries = new List[partitionsNum]; - for (int i = 0; i < partitionsNum; i++) { - partitionEntries[i] = new ArrayList<>(); - } - - List entries; - if (this.isRaw()) { - List rawEntries = this.getRawEntries(); - entries = new ArrayList<>(rawEntries.size()); - for (ByteString byteString : rawEntries) { - Entry entry; - try { - entry = Entry.parseFrom(byteString); - } catch (InvalidProtocolBufferException e) { - throw new RuntimeException(e); - } - entries.add(entry); - } - } else { - entries = this.getEntries(); - } - - for (Entry entry : entries) { - CanalEntry.RowChange rowChange; - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException(e.getMessage(), e); - } - if (rowChange.getIsDdl()) { - partitionEntries[0].add(entry); - } else { - if (rowChange.getRowDatasList() != null && !rowChange.getRowDatasList().isEmpty()) { - String database = entry.getHeader().getSchemaName(); - String table = entry.getHeader().getTableName(); - - String pk = null; - boolean isMatch = false; - - String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, ","); - for (String pkHashConfig : pkHashConfigArray) { - int i = pkHashConfig.lastIndexOf("."); - if (!pkHashConfig.endsWith(".$pk$")) { - // 如果指定了主键 - pk = pkHashConfig.substring(i + 1); - } - pkHashConfig = pkHashConfig.substring(0, i); - - AviaterRegexFilter aviaterRegexFilter = regexFilters.get(pkHashConfig); - if (aviaterRegexFilter == null) { - aviaterRegexFilter = new AviaterRegexFilter(pkHashConfig); - regexFilters.putIfAbsent(pkHashConfig, aviaterRegexFilter); - } - - isMatch = aviaterRegexFilter.filter(database + "." + table); - if (isMatch) { - break; - } - } - - if (!isMatch) { - // 如果都没有匹配，发送到第一个分区 - partitionEntries[0].add(entry); - } - for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - String pkValue = null; - if (pk != null) { - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getName().equalsIgnoreCase(pk)) { - pkValue = column.getValue(); - break; - } - } - } else { - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getIsKey()) { - pkValue = column.getValue(); - break; - } - } - } - if (pkValue != null) { - int hash = pkValue.hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); - } else { - // 没有找到主键 - partitionEntries[0].add(entry); - } - } - } - } - } - - for (int i = 0; i < partitionsNum; i++) { - List entriesTmp = partitionEntries[i]; - if (!entriesTmp.isEmpty()) { - partitionMessages[i] = new Message(this.id, entriesTmp); - } - } - - return partitionMessages; - } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java deleted file mode 100644 index db7750ba..00000000 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java +++ /dev/null @@ -1,124 +0,0 @@ -package com.alibaba.otter.canal.protocol.aviater; - -import java.util.*; - -import org.apache.commons.lang.StringUtils; - -import com.googlecode.aviator.AviatorEvaluator; -import com.googlecode.aviator.Expression; - -/** - * 基于aviater进行tableName正则匹配的过滤算法 - * - * @author jianghang 2012-7-20 下午06:01:34 - */ -public class AviaterRegexFilter { - - private static final String SPLIT = ","; - private static final String PATTERN_SPLIT = "|"; - private static final String FILTER_EXPRESSION = "regex(pattern,target)"; - private static final RegexFunction regexFunction = new RegexFunction(); - private final Expression exp = AviatorEvaluator.compile(FILTER_EXPRESSION, true); - static { - AviatorEvaluator.addFunction(regexFunction); - } - - private static final Comparator COMPARATOR = new StringComparator(); - - final private String pattern; - final private boolean defaultEmptyValue; - - public AviaterRegexFilter(String pattern){ - this(pattern, true); - } - - public AviaterRegexFilter(String pattern, boolean defaultEmptyValue){ - this.defaultEmptyValue = defaultEmptyValue; - List list = null; - if (StringUtils.isEmpty(pattern)) { - list = new ArrayList(); - } else { - String[] ss = StringUtils.split(pattern, SPLIT); - list = Arrays.asList(ss); - } - - // 对pattern按照从长到短的排序 - // 因为 foo|foot 匹配 foot 会出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot - // 的长度不一样 - Collections.sort(list, COMPARATOR); - // 对pattern进行头尾完全匹配 - list = completionPattern(list); - this.pattern = StringUtils.join(list, PATTERN_SPLIT); - } - - public boolean filter(String filtered) { - if (StringUtils.isEmpty(pattern)) { - return defaultEmptyValue; - } - - if (StringUtils.isEmpty(filtered)) { - return defaultEmptyValue; - } - - Map env = new HashMap(); - env.put("pattern", pattern); - env.put("target", filtered.toLowerCase()); - return (Boolean) exp.execute(env); - } - - /** - * 修复正则表达式匹配的问题，因为使用了 oro 的 matches，会出现： - * - * 
-     * foo|foot 匹配 foot 出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot 的长度不一样
-     *
- * - * 因此此类对正则表达式进行了从长到短的排序 - * - * @author zebin.xuzb 2012-10-22 下午2:02:26 - * @version 1.0.0 - */ - private static class StringComparator implements Comparator { - - @Override - public int compare(String str1, String str2) { - if (str1.length() > str2.length()) { - return -1; - } else if (str1.length() < str2.length()) { - return 1; - } else { - return 0; - } - } - } - - /** - * 修复正则表达式匹配的问题，即使按照长度递减排序，还是会出现以下问题： - * - * 
-     * foooo|f.*t 匹配 fooooot 出错，原因是 fooooot 匹配了 foooo 之后，会将 fooo 和数据进行匹配，但是 foooo 的长度和 fooooot 的长度不一样
-     *
- * - * 因此此类对正则表达式进行头尾完全匹配 - * - * @author simon - * @version 1.0.0 - */ - - private List completionPattern(List patterns) { - List result = new ArrayList(); - for (String pattern : patterns) { - StringBuffer stringBuffer = new StringBuffer(); - stringBuffer.append("^"); - stringBuffer.append(pattern); - stringBuffer.append("$"); - result.add(stringBuffer.toString()); - } - return result; - } - - @Override - public String toString() { - return pattern; - } -} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java deleted file mode 100644 index e04f0c4b..00000000 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java +++ /dev/null @@ -1,38 +0,0 @@ -package com.alibaba.otter.canal.protocol.aviater; - -import java.util.Map; - -import org.apache.oro.text.regex.MalformedPatternException; -import org.apache.oro.text.regex.Pattern; -import org.apache.oro.text.regex.PatternCompiler; -import org.apache.oro.text.regex.Perl5Compiler; - -import com.google.common.base.Function; -import com.google.common.collect.MapMaker; -import com.google.common.collect.MigrateMap; - -public class PatternUtils { - @SuppressWarnings("deprecation") - private static Map patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(), - new Function() { - - public Pattern apply(String pattern) { - try { - PatternCompiler pc = new Perl5Compiler(); - return pc.compile(pattern, - Perl5Compiler.CASE_INSENSITIVE_MASK | Perl5Compiler.READ_ONLY_MASK - | Perl5Compiler.SINGLELINE_MASK); - } catch (MalformedPatternException e) { - throw new RuntimeException(e); - } - } - }); - - public static Pattern getPattern(String pattern) { - return patterns.get(pattern); - } - - public static void clear() { - patterns.clear(); - } -} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java deleted file mode 100644 index be084c25..00000000 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java +++ /dev/null @@ -1,31 +0,0 @@ -package com.alibaba.otter.canal.protocol.aviater; - -import java.util.Map; - -import org.apache.oro.text.regex.Perl5Matcher; - -import com.googlecode.aviator.runtime.function.AbstractFunction; -import com.googlecode.aviator.runtime.function.FunctionUtils; -import com.googlecode.aviator.runtime.type.AviatorBoolean; -import com.googlecode.aviator.runtime.type.AviatorObject; - -/** - * 提供aviator regex的代码扩展 - * - * @author jianghang 2012-7-23 上午10:29:23 - */ -public class RegexFunction extends AbstractFunction { - - public AviatorObject call(Map env, AviatorObject arg1, AviatorObject arg2) { - String pattern = FunctionUtils.getStringValue(arg1, env); - String text = FunctionUtils.getStringValue(arg2, env); - Perl5Matcher matcher = new Perl5Matcher(); - boolean isMatch = matcher.matches(text, PatternUtils.getPattern(pattern)); - return AviatorBoolean.valueOf(isMatch); - } - - public String getName() { - return "regex"; - } - -} diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java new file mode 100644 index 00000000..c99df2f4 --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -0,0 +1,405 @@ +package com.alibaba.otter.canal.common; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.commons.lang.StringUtils; + +import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; +import com.alibaba.otter.canal.protocol.CanalEntry; +import com.alibaba.otter.canal.protocol.CanalEntry.Entry; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.protocol.Message; +import com.google.common.base.Function; +import com.google.common.collect.Lists; +import com.google.common.collect.MapMaker; +import com.google.common.collect.MigrateMap; +import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; + +/** + * process MQ Message utils + * + * @author agapple 2018年12月11日 下午1:28:32 + */ +public class MQMessageUtils { + + @SuppressWarnings("deprecation") + private static Map> partitionDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(), + new Function>() { + + public List apply(String pkHashConfigs) { + List datas = Lists.newArrayList(); + String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, + ","); + // schema.table#id^name + for (String pkHashConfig : pkHashConfigArray) { + PartitionData data = new PartitionData(); + int i = pkHashConfig.lastIndexOf("#"); + if (i > 0) { + data.pkNames = Lists.newArrayList(StringUtils.split(pkHashConfig.substring(i + 1), + '^')); + pkHashConfig = pkHashConfig.substring(0, + i); + } + if (!isWildCard(pkHashConfig)) { + data.simpleName = pkHashConfig; + } else { + data.regexFilter = new AviaterRegexFilter(pkHashConfig); + } + datas.add(data); + } + + return datas; + } + }); + + /** + * 将 message 分区 + * + * @param partitionsNum 分区数 + * @param pkHashConfigs 分区库表主键正则表达式 + * @return 分区message数组 + */ + @SuppressWarnings("unchecked") + public static Message[] messagePartition(Message message, Integer partitionsNum, String pkHashConfigs) { + if (partitionsNum == null) { + partitionsNum = 1; + } + Message[] partitionMessages = new Message[partitionsNum]; + List[] partitionEntries = new List[partitionsNum]; + for (int i = 0; i < partitionsNum; i++) { + partitionEntries[i] = new ArrayList<>(); + } + + List entries; + if (message.isRaw()) { + List rawEntries = message.getRawEntries(); + entries = new ArrayList<>(rawEntries.size()); + for (ByteString byteString : rawEntries) { + Entry entry; + try { + entry = Entry.parseFrom(byteString); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + entries.add(entry); + } + } else { + entries = message.getEntries(); + } + + for (Entry entry : entries) { + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + + if (rowChange.getIsDdl()) { + partitionEntries[0].add(entry); + } else { + if (rowChange.getRowDatasList() != null && !rowChange.getRowDatasList().isEmpty()) { + String database = entry.getHeader().getSchemaName(); + String table = entry.getHeader().getTableName(); + List pkNames = getParitionHashColumns(database + "." + table, pkHashConfigs); + + if (pkNames == null) { + // 如果都没有匹配，发送到第一个分区 + partitionEntries[0].add(entry); + } else { + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + int hashCode = table.hashCode(); + if (pkNames.isEmpty()) { + // isEmpty use default pkNames + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + hashCode = hashCode ^ column.getValue().hashCode(); + } + } + } else { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (checkPkNamesHasContain(pkNames, column.getName())) { + hashCode = hashCode ^ column.getValue().hashCode(); + } + } + } + + int pkHash = Math.abs(hashCode) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + } + } + } + } + } + + for (int i = 0; i < partitionsNum; i++) { + List entriesTmp = partitionEntries[i]; + if (!entriesTmp.isEmpty()) { + partitionMessages[i] = new Message(message.getId(), entriesTmp); + } + } + + return partitionMessages; + } + + /** + * 将Message转换为FlatMessage + * + * @param message 原生message + * @return FlatMessage列表 + */ + public static List messageConverter(Message message) { + try { + if (message == null) { + return null; + } + + List flatMessages = new ArrayList<>(); + List entrys = null; + if (message.isRaw()) { + List rawEntries = message.getRawEntries(); + entrys = new ArrayList(rawEntries.size()); + for (ByteString byteString : rawEntries) { + CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); + entrys.add(entry); + } + } else { + entrys = message.getEntries(); + } + + for (CanalEntry.Entry entry : entrys) { + if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN + || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { + continue; + } + + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + + entry.toString(), e); + } + + CanalEntry.EventType eventType = rowChange.getEventType(); + + FlatMessage flatMessage = new FlatMessage(message.getId()); + flatMessages.add(flatMessage); + flatMessage.setDatabase(entry.getHeader().getSchemaName()); + flatMessage.setTable(entry.getHeader().getTableName()); + flatMessage.setIsDdl(rowChange.getIsDdl()); + flatMessage.setType(eventType.toString()); + flatMessage.setEs(entry.getHeader().getExecuteTime()); + flatMessage.setTs(System.currentTimeMillis()); + flatMessage.setSql(rowChange.getSql()); + + if (!rowChange.getIsDdl()) { + Map sqlType = new LinkedHashMap<>(); + Map mysqlType = new LinkedHashMap<>(); + List> data = new ArrayList<>(); + List> old = new ArrayList<>(); + + Set updateSet = new HashSet<>(); + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE + && eventType != CanalEntry.EventType.DELETE) { + continue; + } + + Map row = new LinkedHashMap<>(); + List columns; + + if (eventType == CanalEntry.EventType.DELETE) { + columns = rowData.getBeforeColumnsList(); + } else { + columns = rowData.getAfterColumnsList(); + } + + for (CanalEntry.Column column : columns) { + if (column.getIsKey()) { + flatMessage.addPkName(column.getName()); + } + sqlType.put(column.getName(), column.getSqlType()); + mysqlType.put(column.getName(), column.getMysqlType()); + if (column.getIsNull()) { + row.put(column.getName(), null); + } else { + row.put(column.getName(), column.getValue()); + } + // 获取update为true的字段 + if (column.getUpdated()) { + updateSet.add(column.getName()); + } + } + if (!row.isEmpty()) { + data.add(row); + } + + if (eventType == CanalEntry.EventType.UPDATE) { + Map rowOld = new LinkedHashMap<>(); + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + if (updateSet.contains(column.getName())) { + if (column.getIsNull()) { + rowOld.put(column.getName(), null); + } else { + rowOld.put(column.getName(), column.getValue()); + } + } + } + // update操作将记录修改前的值 + if (!rowOld.isEmpty()) { + old.add(rowOld); + } + } + } + if (!sqlType.isEmpty()) { + flatMessage.setSqlType(sqlType); + } + if (!mysqlType.isEmpty()) { + flatMessage.setMysqlType(mysqlType); + } + if (!data.isEmpty()) { + flatMessage.setData(data); + } + if (!old.isEmpty()) { + flatMessage.setOld(old); + } + } + } + return flatMessages; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + /** + * 将FlatMessage按指定的字段值hash拆分 + * + * @param flatMessage flatMessage + * @param partitionsNum 分区数量 + * @param pkHashConfigs hash映射 + * @return 拆分后的flatMessage数组 + */ + public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs) { + if (partitionsNum == null) { + partitionsNum = 1; + } + FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; + + if (flatMessage.getIsDdl()) { + partitionMessages[0] = flatMessage; + } else { + if (flatMessage.getData() != null && !flatMessage.getData().isEmpty()) { + String database = flatMessage.getDatabase(); + String table = flatMessage.getTable(); + List pkNames = getParitionHashColumns(database + "." + table, pkHashConfigs); + if (pkNames == null) { + // 如果都没有匹配，发送到第一个分区 + partitionMessages[0] = flatMessage; + } else { + if (pkNames.isEmpty()) { + pkNames = flatMessage.getPkNames(); + } + + int hashCode = table.hashCode(); + int idx = 0; + for (Map row : flatMessage.getData()) { + for (String pkName : pkNames) { + String value = row.get(pkName); + if (value == null) { + value = ""; + } + hashCode = hashCode ^ value.hashCode(); + } + + int pkHash = Math.abs(hashCode) % partitionsNum; + // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 + pkHash = Math.abs(pkHash); + + FlatMessage flatMessageTmp = partitionMessages[pkHash]; + if (flatMessageTmp == null) { + flatMessageTmp = new FlatMessage(flatMessage.getId()); + partitionMessages[pkHash] = flatMessageTmp; + flatMessageTmp.setDatabase(flatMessage.getDatabase()); + flatMessageTmp.setTable(flatMessage.getTable()); + flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); + flatMessageTmp.setType(flatMessage.getType()); + flatMessageTmp.setSql(flatMessage.getSql()); + flatMessageTmp.setSqlType(flatMessage.getSqlType()); + flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); + flatMessageTmp.setEs(flatMessage.getEs()); + flatMessageTmp.setTs(flatMessage.getTs()); + } + List> data = flatMessageTmp.getData(); + if (data == null) { + data = new ArrayList<>(); + flatMessageTmp.setData(data); + } + data.add(row); + if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { + List> old = flatMessageTmp.getOld(); + if (old == null) { + old = new ArrayList<>(); + flatMessageTmp.setOld(old); + } + old.add(flatMessage.getOld().get(idx)); + } + idx++; + } + } + } + } + return partitionMessages; + } + + /** + * match return List , not match return null + */ + public static List getParitionHashColumns(String name, String pkHashConfigs) { + List datas = partitionDatas.get(pkHashConfigs); + for (PartitionData data : datas) { + if (data.simpleName != null) { + if (data.simpleName.equalsIgnoreCase(name)) { + return data.pkNames; + } + } else { + if (data.regexFilter.filter(name)) { + return data.pkNames; + } + } + } + + return null; + } + + public static boolean checkPkNamesHasContain(List pkNames, String name) { + for (String pkName : pkNames) { + if (pkName.equalsIgnoreCase(name)) { + return true; + } + } + + return false; + } + + private static boolean isWildCard(String value) { + // not contaiins '.' ? + return StringUtils.containsAny(value, new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$', + '^' }); + } + + public static class PartitionData { + + public String simpleName; + public AviaterRegexFilter regexFilter; + public List pkNames = Lists.newArrayList(); + } + +} diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 7e82f9c6..6f6122ec 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -1,6 +1,5 @@ package com.alibaba.otter.canal.common; -import java.util.Map; /** * kafka 配置项 diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 0dd93ea5..91d42f24 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -12,6 +12,7 @@ import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; +import com.alibaba.otter.canal.common.MQMessageUtils; import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; @@ -88,7 +89,8 @@ public class CanalKafkaProducer implements CanalMQProducer { message); } else { if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { - Message[] messages = message.messagePartition(canalDestination.getPartitionsNum(), + Message[] messages = MQMessageUtils.messagePartition(message, + canalDestination.getPartitionsNum(), canalDestination.getPartitionHash()); int length = messages.length; for (int i = 0; i < length; i++) { @@ -119,13 +121,12 @@ public class CanalKafkaProducer implements CanalMQProducer { } } else { // 发送扁平数据json - List flatMessages = FlatMessage.messageConverter(message); + List flatMessages = MQMessageUtils.messageConverter(message); if (flatMessages != null) { for (FlatMessage flatMessage : flatMessages) { if (canalDestination.getPartition() != null) { try { - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), + ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); @@ -139,7 +140,7 @@ public class CanalKafkaProducer implements CanalMQProducer { } else { if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { - FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage, + FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, canalDestination.getPartitionsNum(), canalDestination.getPartitionHash()); int length = partitionFlatMessage.length; @@ -147,8 +148,7 @@ public class CanalKafkaProducer implements CanalMQProducer { FlatMessage flatMessagePart = partitionFlatMessage[i]; if (flatMessagePart != null) { try { - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), + ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), i, null, JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); @@ -163,8 +163,7 @@ public class CanalKafkaProducer implements CanalMQProducer { } } else { try { - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), + ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), 0, null, JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index 56648d4e..647fdfcf 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -16,6 +16,7 @@ import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.CanalMessageSerializer; +import com.alibaba.otter.canal.common.MQMessageUtils; import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.server.exception.CanalServerException; @@ -61,8 +62,8 @@ public class CanalRocketMQProducer implements CanalMQProducer { if (!mqProperties.getFlatMessage()) { try { if (destination.getPartition() != null) { - Message message = new Message(destination.getTopic(), - CanalMessageSerializer.serializer(data, mqProperties.isFilterTransactionEntry())); + Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data, + mqProperties.isFilterTransactionEntry())); if (logger.isDebugEnabled()) { logger.debug("send message:{} to destination:{}, partition: {}", message, @@ -82,8 +83,9 @@ public class CanalRocketMQProducer implements CanalMQProducer { }, null); } else { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { - com.alibaba.otter.canal.protocol.Message[] messages = data - .messagePartition(destination.getPartitionsNum(), destination.getPartitionHash()); + com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(data, + destination.getPartitionsNum(), + destination.getPartitionHash()); int length = messages.length; for (int i = 0; i < length; i++) { com.alibaba.otter.canal.protocol.Message dataPartition = messages[i]; @@ -125,7 +127,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { return; } } else { - List flatMessages = FlatMessage.messageConverter(data); + List flatMessages = MQMessageUtils.messageConverter(data); if (flatMessages != null) { for (FlatMessage flatMessage : flatMessages) { if (destination.getPartition() != null) { @@ -136,8 +138,8 @@ public class CanalRocketMQProducer implements CanalMQProducer { destination.getTopic(), destination.getPartition()); } - Message message = new Message(destination.getTopic(), - JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes()); + Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessage, + SerializerFeature.WriteMapNullValue).getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override @@ -152,7 +154,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } } else { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { - FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage, + FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, destination.getPartitionsNum(), destination.getPartitionHash()); int length = partitionFlatMessage.length; @@ -172,17 +174,16 @@ public class CanalRocketMQProducer implements CanalMQProducer { this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override - public MessageQueue select(List mqs, Message msg, - Object arg) { + public MessageQueue select(List mqs, Message msg, Object arg) { if (index > mqs.size()) { - throw new CanalServerException( - "partition number is error,config num:" + throw new CanalServerException("partition number is error,config num:" + destination.getPartitionsNum() + ", mq num: " + mqs.size()); } return mqs.get(index); } - }, null); + }, + null); } catch (Exception e) { logger.error("send flat message to hashed partition error", e); callback.rollback(); From 48de904ee169e1f9343026aedb82bdfd95cead15 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 11 Dec 2018 14:43:13 +0800 Subject: [PATCH 15/15] =?UTF-8?q?=E5=8E=BB=E6=8E=89=20#1266=20=E4=B8=AD?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E7=9A=84=20protocol=20=E7=9A=84=E5=A4=9A?= =?UTF-8?q?=E4=BD=99=E7=9A=84=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- protocol/pom.xml | 8 -------- 1 file changed, 8 deletions(-) diff --git a/protocol/pom.xml b/protocol/pom.xml index 0f7904d6..1b776b7e 100644 --- a/protocol/pom.xml +++ b/protocol/pom.xml @@ -26,13 +26,5 @@ commons-lang commons-lang - - oro - oro - - - com.googlecode.aviator - aviator -