Compare commits

..

104 Commits

Author SHA1 Message Date
agapple
5f4bbab722 [maven-release-plugin] prepare release canal-1.0.23 2016-12-30 14:57:07 +08:00
agapple
a483f03711 fixed #227 , loadBinlogChecksum NPE 2016-12-30 14:51:22 +08:00
agapple
a1207060bf format 2016-12-30 14:32:52 +08:00
agapple
5055fdb556 Merge pull request #245 from dangxia/master
issues#244
2016-12-30 14:30:00 +08:00
XuehuiHe
a69b69c724 issues#244 2016-12-30 11:39:02 +08:00
agapple
90e6d34fb4 Merge pull request #243 from deeplinux/master
原来的example内的startup.bat脚本直接运行时报语法错误
2016-12-29 13:45:32 +08:00
deeplinux
907ecd43e4 修改批处理中的!=为NEQ 2016-12-28 08:06:18 +08:00
deeplinux
c57a921a1a 修正批处理中的语法错误 2016-12-27 17:57:07 +08:00
agapple
83fee4ffbb Merge pull request #231 from alexwangx/alexwang
fixed issue #227 , fixed Mysql5.1.73 throw NullPointerException
2016-12-02 18:06:36 +08:00
alexwang
bc7927fe75 fixed issue #227 , fixed Mysql5.1.73 throw NullPointerException 2016-12-01 17:39:51 +08:00
agapple
e67f27018c fixed support json inteAddress 2016-11-28 14:11:12 +08:00
agapple
3f38ad6f49 fixed issue #227 , fixed binlog_checksum variables 2016-11-28 14:10:57 +08:00
agapple
96dca0b9e6 fixed issue #136 , support rds no primary key && mysql.ha_health_check 2016-11-17 17:52:52 +08:00
agapple
ad2a96b3f1 fixed json type cast exception 2016-11-10 11:12:35 +08:00
agapple
6284a429a0 fixed lazy 2016-10-11 21:37:39 +08:00
agapple
83d8a39d16 add ServerRunningMonitor.init 2016-10-11 19:39:55 +08:00
agapple
4898829384 Merge pull request #213 from lulu2panpan/master
修复issue #212
2016-10-11 19:25:06 +08:00
agapple
69a7e6d083 Merge pull request #208 from jianan789/master
mysql5.6.29'@@global.binlog_checksum'带单引号导致master退出,优化binlogFormat判断
2016-10-11 19:16:34 +08:00
agapple
a5d6c3b747 Merge pull request #210 from xujianhai/patch-1
不好的设计
2016-10-11 19:14:22 +08:00
agapple
86378cad32 Merge pull request #209 from lcybo/master
Fix a minor fault: Canal instance reload again after startup.
2016-10-11 19:11:30 +08:00
lubiao
8626c6ee24 1、SessionHandler类中执行SUBSCRIPTION时,应该先尝试启动runningMonitor再调用embeddedServer的subscribe方法
2、embeddedServer的subscribe方法中应该首先进行checkStart操作,否则可能引起metaManager的脏启动
3、SessionHandler中增加日志
2016-10-11 18:46:30 +08:00
lubiao
350f8637c3 1、BinLogFileQueue的listBinlogFiles方法中对binlog文件名进行更严格的验证,必须以数字结尾
2、LocalBinLogConnection类中解决如下bug
	2.1第二个dump方法中增加代码"decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);",解决"((QueryLogEvent)event).getQuery()"语句出现乱码的问题(BEGIN后面多了4个字节),出现乱码不会进入if语句,导致找到的binlog位点不准确
	2.2第一个dump方法中,被注释的decode代码才是正确的,和DirectLogFetcher不同,此处需要构造子循环进行遍历
	2.3第一个dump方法中,将每个binlog-file的首个event设置为RotateLogEvent,否则会导致所有event的binlogFileName都是默认的mysql-bin.000001。这么改完之后,和DirectLogFetcher的行为相契合,后者在每个binlogfile的首个event就是RotateLogEvent
3、LocalBinlogEventParser中增加元数据处理的相关代码,不加的话,所有的event都不带列名,基本上没法儿使用
2016-10-11 17:28:24 +08:00
李建安
96454d4387 修复开启checksum后rotateLogEvent数据包解析时binlog name乱码 2016-10-10 18:30:09 +08:00
徐建海
8a293e4b0c Update AbstractEventParser.java 2016-10-06 19:13:50 +08:00
徐建海
006dd63646 不好的设计 2016-10-06 15:48:05 +08:00
Chuanyi Li
b17c11868c Fix a minor fault: Canal instance reload again after startup. 2016-10-02 13:43:17 +08:00
李建安
0c9eecd758 mysql5.6.29'@@global.binlog_checksum'带单引号导致master退出,优化binlogFormat判断 2016-09-29 17:21:59 +08:00
agapple
5d9d5b1e44 fixed testcase 2016-09-29 11:43:35 +08:00
agapple
b23e814c48 Merge pull request #203 from kevinKaiF/master
fixed issue # 201
2016-09-29 10:31:29 +08:00
zikaifeng
b349b400cf 修复issue # 201 2016-09-27 16:22:48 +08:00
agapple
c86fba6092 Merge pull request #199 from MarkLinHz/master
upgrade protobuf version
2016-09-19 14:03:20 +08:00
yinxiu
9742fdc9d9 upgrade protobuf version 2016-09-19 12:24:06 +08:00
agapple
c99b4b8cd1 fixed NPE #200 2016-09-07 12:37:21 +08:00
agapple
aa8ea96572 fixed issue #195 2016-08-26 18:25:52 +08:00
agapple
f6cf88a6bd support 5.7 2016-08-04 22:53:59 +08:00
agapple
14a5762fbb support 5.7 2016-08-04 22:53:22 +08:00
agapple
eda24df3fe fixed kill CONNECTION exception 2016-07-26 10:59:09 +08:00
agapple
3a5a4b06a9 [maven-release-plugin] prepare for next development iteration 2016-06-30 16:59:17 +08:00
agapple
ba173fcd48 [maven-release-plugin] prepare release canal-1.0.22 2016-06-30 16:59:05 +08:00
agapple
69170f58df fixed issue #172, support rds 2016-06-30 16:45:42 +08:00
agapple
c3e38e50e2 fixed issue #187 ,support MySQL 5.7 Json 2016-06-30 16:23:05 +08:00
agapple
ae05a3f39f fixed issue #172, 精确识别主备切换,尝试基于时间戳重新定位位点,确保不丢数据 2016-06-28 22:58:55 +08:00
agapple
6eafc73a65 Merge pull request #177 from chenyishun/master
RDS主从库切换时候异常处理
2016-05-10 18:06:11 +08:00
Yishun.Chen
63b3c4ab54 IOException限制retry次数,超过次数后跳过binlog解析 2016-05-03 16:30:38 +08:00
Yishun.Chen
ecba855bf9 IOException限制retry次数,超过次数后跳过binlog解析 2016-05-03 16:30:01 +08:00
Yishun.Chen
b599846470 IOException限制retry次数,超过次数后跳过binlog解析 2016-05-03 16:29:23 +08:00
agapple
8aefd7e79b fixed issue #171, getNextAddress 2016-04-29 17:44:06 +08:00
agapple
4e76a4bd6e fixed issue #176 , handle running deleted 2016-04-29 17:43:16 +08:00
agapple
f1a1cb26cc fixed issue #171, compiler error 2016-04-29 11:29:16 +08:00
agapple
4f1e3bafe7 fixed issue #171 , client HA bug , modify SimpleCanalConnector 2016-04-29 11:26:37 +08:00
agapple
6a03ffaf7b Merge pull request #173 from lulu2panpan/master
Canal客户端高可用bug修复
2016-04-29 11:16:24 +08:00
lulu2panpan
d7d969915e Update ClientRunningMonitor.java 2016-04-29 09:50:37 +08:00
lulu2panpan
4898ed7b83 Update SimpleCanalConnector.java 2016-04-29 09:40:27 +08:00
lulu2panpan
11e1d1d600 Update CanalConnectors.java 2016-04-29 09:31:40 +08:00
lulu2panpan
e614c6141d Update ClusterCanalConnector.java 2016-04-29 09:28:48 +08:00
lulu2panpan
8ded94a09e Update ClusterCanalConnector.java 2016-04-29 09:28:12 +08:00
agapple
38cc81b7da fixed issue #162 , fixed IDE CanalLauncher 2016-04-07 11:25:59 +08:00
agapple
533c5cb8f6 fixed issue #119 , fixed batch data changeColumns 2016-04-05 23:42:08 +08:00
agapple
5932e1e31f edit url 2016-03-05 18:46:21 +08:00
agapple
70dda221f1 add yugong 2016-03-05 18:44:57 +08:00
agapple
c32a5d0882 fixed issue #139 , KILL CONNECTION binlogdump thread 2016-03-01 22:19:02 +08:00
agapple
fbd99acac2 Merge pull request #145 from OpenZYK/master
添加rowdata过滤
2016-02-25 18:14:20 +08:00
zhangyukun
51cfa7fa96 Update AbstractMysqlEventParser.java 2016-02-25 13:56:07 +08:00
zhangyukun
652101035e 添加Rows过滤 2016-02-25 11:44:22 +08:00
zhangyukun
1813c1eff9 Update canal.properties 2016-02-25 11:39:10 +08:00
zhangyukun
31ea9fa8f0 新增canal.instance.filter.rows过滤 2016-02-25 11:38:09 +08:00
zhangyukun
369acd8eed 新增canal.instance.filter.rows过滤 2016-02-25 11:37:45 +08:00
zhangyukun
891eb960ca 新增canal.instance.filter.rows过滤 2016-02-25 11:37:17 +08:00
zhangyukun
2e5a165b88 新增canal.instance.filter.rows过滤
为parseRowsEvent方法添加过滤
2016-02-25 11:35:46 +08:00
zhangyukun
6b9c1305ff 添加dml过滤判断
修复parseRowsEvent方法中没有dml过滤的判断而导致dml操作的数据总是会输出到store中的bug
2016-02-24 21:18:01 +08:00
agapple
4f504aab31 fixed testcase compiler error 2016-02-24 17:05:14 +08:00
agapple
dc28c68bf0 fixed format 2016-02-23 17:05:12 +08:00
agapple
38f2f98da3 Merge pull request #144 from dingjun84/master
注销父类已经有的属性,避免super.start()调用了没有在子类初始化的同名属性
2016-02-23 17:03:20 +08:00
dingjun
efebbe301a 注销父类已经有的属性,避免super.start()调用了没有在子类初始化的同名属性 2016-02-23 10:40:42 +08:00
agapple
bf10b16138 migrate 2016-02-04 11:22:36 +08:00
agapple
37478f3409 Merge pull request #138 from MarkLinHz/master
make canal server become a single instance
2016-02-04 11:02:23 +08:00
yinxiu
d2f98cda87 remove duplicate code 2016-01-14 10:59:40 +08:00
yinxiu
ed74d934d3 add setAlarmHandler for otter 2016-01-11 16:15:41 +08:00
yinxiu
7b29828f60 add AbstractCanalInstance and remove duplicate code 2016-01-08 20:32:08 +08:00
yinxiu
9164fe4ca7 make canal server become a single instance 2016-01-07 17:23:58 +08:00
agapple
51ac311aa1 fixed git ignore 2015-12-31 14:31:55 +08:00
agapple
643bd5cb71 [maven-release-plugin] prepare for next development iteration 2015-12-30 16:23:30 +08:00
agapple
8375ea4b7d [maven-release-plugin] prepare release canal-1.0.21 2015-12-30 16:23:20 +08:00
agapple
72ceb58fa3 fixed issue #135 , isUpdate bugfix 2015-12-02 14:37:06 +08:00
agapple
44f2b74862 fixed issue #130 , support mysql5.6 time2/timestamp2/datetime2 javaType 2015-11-13 14:16:12 +08:00
agapple
5a34195201 Merge pull request #128 from likeloveluck/simple_ddl_pattern
fix rename table ddl parser tablename and schemaname error
2015-09-07 15:49:12 +08:00
jiawei
d976932177 fix rename table ddl parser tablename and schemaname error 2015-09-01 12:17:41 +08:00
agapple
d887c8f763 Merge pull request #127 from dozer47528/master
fix not skip filter and read the error data
2015-08-04 11:24:14 +08:00
Dozer
197a1446b7 fix not skip filter and read the error data 2015-07-31 17:18:42 +08:00
agapple
155c00b256 Merge pull request #124 from smilingleo/master
fixed a hard coded mysql host ip address
2015-07-28 13:51:44 +08:00
Leo
b16f4844a1 fixed a hard coded mysql host ip address 2015-07-23 10:42:03 +08:00
agapple
f2f2bfb538 add rename testcase 2015-07-17 14:43:15 +08:00
agapple
5f6580399c fixed format 2015-07-17 14:41:39 +08:00
agapple
10930b9e3d Merge pull request #122 from likeloveluck/rename_bug_fix
修复rename table命令中包含多个表,解析失败问题
2015-07-17 14:33:38 +08:00
jiawei
d1afeb7540 change result reference 2015-07-16 19:25:46 +08:00
jiawei
119af9fca1 change rename operate 2015-07-15 19:20:30 +08:00
agapple
f0ce4b569e Merge pull request #121 from huyanping/patch-1
Update README.md
2015-07-14 16:59:24 +08:00
Jenner
4459b8acdf Update README.md 2015-07-14 14:42:13 +08:00
agapple
bb5d5749bd Merge pull request #120 from smilingleo/master
增加了一个脚本来自动获取mysql master binlog信息,并自动更新instance配置文件
2015-07-05 11:10:56 +08:00
smilingleo
169baafb8b correct format 2015-07-03 10:32:40 +08:00
smilingleo
199f30de87 add a script to fetch mysql master binlog file and sync position automatically 2015-07-03 10:29:58 +08:00
smilingleo
87c14f0b51 Revert "test"
This reverts commit f35d965f63.
2015-07-03 10:08:03 +08:00
smilingleo
f35d965f63 test 2015-07-03 10:02:10 +08:00
agapple
9956b3557f [maven-release-plugin] prepare for next development iteration 2015-06-30 10:23:54 +08:00
73 changed files with 13265 additions and 4685 deletions

1
.gitignore vendored
View File

@@ -13,3 +13,4 @@ otter.properties
jtester.properties
.idea/
*.iml
.DS_Store

View File

@@ -9,7 +9,7 @@
<h1>背景</h1>
<p style="font-size: 14px;"> 早期,阿里巴巴B2B公司因为存在杭州和美国双机房部署,存在跨机房同步的业务需求。不过早期的数据库同步业务,主要是基于trigger的方式获取增量变更,不过从2010年开始,阿里系公司开始逐步的尝试基于数据库的日志解析,获取增量变更进行同步,由此衍生出了增量订阅&amp;消费的业务,从此开启了一段新纪元。</p>
<p style="font-size: 14px;"> ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析,当前的canal开源版本支持5.6及以下的版本(阿里内部mysql 5.6.10, mysql 5.5.18和5.1.40/48)</p>
<p style="font-size: 14px;"> ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析,当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)</p>
<p style="font-size: 14px;"> </p>
<p style="font-size: 14px;">基于日志增量订阅&amp;消费支持的业务:</p>
<ol style="font-size: 14px;">
@@ -36,7 +36,7 @@
<li>slave重做中继日志中的事件,将改变反映它自己的数据。</li>
</ol>
<h3>canal的工作原理:</h3>
<p><img width="590" src="http://dl.iteye.com/upload/attachment/0080/3107/c87b67ba-394c-3086-9577-9db05be04c95.jpg" alt="" height="273">
<p><img width="590" src="https://camo.githubusercontent.com/46c626b4cde399db43b2634a7911a04aecf273a0/687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333130372f63383762363762612d333934632d333038362d393537372d3964623035626530346339352e6a7067" alt="" height="273">
<p>原理相对比较简单:</p>
<ol>
<li>canal模拟mysql slave的交互协议,伪装自己为mysql slave,向mysql master发送dump协议</li>
@@ -62,11 +62,16 @@ See the wiki page for : <a href="https://github.com/alibaba/canal/wiki" >wiki文
<li><a href="https://github.com/alibaba/canal/releases">Download</a></li>
</ul>
<h1>相关资料: </h1>
<h1>相关资料</h1>
* ADC阿里技术嘉年华分享ppt (放在google docs上,可能需要翻墙): <a href="https://docs.google.com/presentation/d/1MkszUPYRDkfVPz9IqOT1LLT5d9tuwde_WC8GZvjaDRg/edit?usp=sharing">ppt下载</href>
* [与阿里巴巴的RocketMQ配合使用](https://github.com/alibaba/RocketMQ)
<h1>相关开源</h1>
<ol>
<li>阿里巴巴分布式数据库同步系统(解决中美异地机房):<a href="http://github.com/alibaba/otter">http://github.com/alibaba/otter</a></li>
<li>阿里巴巴去Oracle数据迁移同步工具(目标支持MySQL/DRDS):<a href="http://github.com/alibaba/yugong">http://github.com/alibaba/yugong</a></li>
</ol>
<h1>问题反馈</h1>
<ol>

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -10,5 +10,7 @@ import java.net.SocketAddress;
*/
public interface CanalNodeAccessStrategy {
SocketAddress currentNode();
SocketAddress nextNode();
}

View File

@@ -41,11 +41,17 @@ public class ClusterCanalConnector implements CanalConnector {
public void connect() throws CanalClientException {
while (currentConnector == null) {
SocketAddress nextAddress = this.accessStrategy.nextNode();
int times = 0;
while (true) {
try {
currentConnector = new SimpleCanalConnector(nextAddress, username, password, destination);
currentConnector = new SimpleCanalConnector(null, username, password, destination) {
@Override
public SocketAddress getNextAddress() {
return accessStrategy.nextNode();
}
};
currentConnector.setSoTimeout(soTimeout);
if (filter != null) {
currentConnector.setFilter(filter);
@@ -57,7 +63,7 @@ public class ClusterCanalConnector implements CanalConnector {
currentConnector.connect();
break;
} catch (Exception e) {
logger.warn("failed to connect to:{} after retry {} times", nextAddress, times);
logger.warn("failed to connect to:{} after retry {} times", accessStrategy.currentNode(), times);
currentConnector.disconnect();
currentConnector = null;
// retry for #retryTimes for each node when trying to

View File

@@ -62,6 +62,11 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
initRunning(this.zkClient.readData(runningPath, true));
}
@Override
public SocketAddress currentNode() {
return nextNode();
}
public SocketAddress nextNode() {
if (runningAddress != null) {// 如果服务已经启动,直接选择当前正在工作的节点
return runningAddress;
@@ -109,4 +114,5 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
return zkClient;
}
}

View File

@@ -125,6 +125,10 @@ public class SimpleCanalConnector implements CanalConnector {
try {
channel = SocketChannel.open();
channel.socket().setSoTimeout(soTimeout);
SocketAddress address = getAddress();
if (address == null) {
address = getNextAddress();
}
channel.connect(address);
Packet p = Packet.parseFrom(readNextPacket(channel));
if (p.getVersion() != 1) {
@@ -434,6 +438,10 @@ public class SimpleCanalConnector implements CanalConnector {
}
}
public SocketAddress getNextAddress() {
return null;
}
public SocketAddress getAddress() {
return address;
}

View File

@@ -32,4 +32,9 @@ public class SimpleNodeAccessStrategy implements CanalNodeAccessStrategy {
}
}
@Override
public SocketAddress currentNode() {
return nodes.get(index);
}
}

View File

@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.client.impl.running;
import java.net.InetSocketAddress;
import java.text.MessageFormat;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -20,6 +21,7 @@ import com.alibaba.otter.canal.common.utils.BooleanMutex;
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.protocol.exception.CanalClientException;
/**
* clinet running控制
@@ -62,6 +64,8 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
public void handleDataDeleted(String dataPath) throws Exception {
MDC.put("destination", destination);
mutex.set(false);
// 触发一下退出,可能是人为干预的释放操作或者网络闪断引起的session expired timeout
processActiveExit();
if (!release && activeData != null && isMine(activeData.getAddress())) {
// 如果上一次active的状态就是本机,则即时触发一下active抢占
initRunning();
@@ -96,7 +100,11 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
releaseRunning(); // 尝试一下release
}
public void initRunning() {
// 改动记录:
// 1,在方法上加synchronized关键字,保证同步顺序执行;
// 2,判断Zk上已经存在的activeData是否是本机,是的话把mutex重置为true,否则会导致死锁
// 3,增加异常处理,保证出现异常时,running节点能被删除,否则会导致死锁
public synchronized void initRunning() {
if (!isStart()) {
return;
}
@@ -116,11 +124,22 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
initRunning();
} else {
activeData = JsonUtils.unmarshalFromByte(bytes, ClientRunningData.class);
// 如果发现已经存在,判断一下是否自己,避免活锁
if (activeData.getAddress().contains(":") && isMine(activeData.getAddress())) {
mutex.set(true);
}
}
} catch (ZkNoNodeException e) {
zkClient.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientData.getClientId()),
true); // 尝试创建父节点
initRunning();
} catch (Throwable t) {
logger.error(MessageFormat.format("There is an error when execute initRunning method, with destination [{0}].",
destination),
t);
// 出现任何异常尝试release
releaseRunning();
throw new CanalClientException("something goes wrong in initRunning method. ", t);
}
}

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>canal.common</artifactId>

View File

@@ -1,12 +1,19 @@
package com.alibaba.otter.canal.common.utils;
import java.io.IOException;
import java.lang.reflect.Type;
import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.List;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.serializer.JSONSerializer;
import com.alibaba.fastjson.serializer.ObjectSerializer;
import com.alibaba.fastjson.serializer.PropertyFilter;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializeWriter;
import com.alibaba.fastjson.serializer.SerializerFeature;
@@ -17,6 +24,12 @@ import com.alibaba.fastjson.serializer.SerializerFeature;
*/
public class JsonUtils {
static {
SerializeConfig.getGlobalInstance().put(InetAddress.class, InetAddressSerializer.instance);
SerializeConfig.getGlobalInstance().put(Inet4Address.class, InetAddressSerializer.instance);
SerializeConfig.getGlobalInstance().put(Inet6Address.class, InetAddressSerializer.instance);
}
public static <T> T unmarshalFromByte(byte[] bytes, Class<T> targetClass) {
return (T) JSON.parseObject(bytes, targetClass);// 默认为UTF-8
}
@@ -70,4 +83,21 @@ public class JsonUtils {
out.close();
}
}
public static class InetAddressSerializer implements ObjectSerializer {
public static InetAddressSerializer instance = new InetAddressSerializer();
public void write(JSONSerializer serializer, Object object, Object fieldName, Type fieldType)
throws IOException {
if (object == null) {
serializer.writeNull();
return;
}
InetAddress address = (InetAddress) object;
// 优先使用name
serializer.write(address.getHostName());
}
}
}

View File

@@ -87,6 +87,10 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
}
public void init() {
processStart();
}
public void start() {
super.start();
processStart();

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -0,0 +1,402 @@
package com.taobao.tddl.dbsync.binlog;
/**
* 处理下MySQL json二进制转化为可读的字符串
*
* @author agapple 2016年6月30日 上午11:26:17
* @since 1.0.22
*/
public class JsonConversion {
// JSON TYPE
public static final int JSONB_TYPE_SMALL_OBJECT = 0x0;
public static final int JSONB_TYPE_LARGE_OBJECT = 0x1;
public static final int JSONB_TYPE_SMALL_ARRAY = 0x2;
public static final int JSONB_TYPE_LARGE_ARRAY = 0x3;
public static final int JSONB_TYPE_LITERAL = 0x4;
public static final int JSONB_TYPE_INT16 = 0x5;
public static final int JSONB_TYPE_UINT16 = 0x6;
public static final int JSONB_TYPE_INT32 = 0x7;
public static final int JSONB_TYPE_UINT32 = 0x8;
public static final int JSONB_TYPE_INT64 = 0x9;
public static final int JSONB_TYPE_UINT64 = 0xA;
public static final int JSONB_TYPE_DOUBLE = 0xB;
public static final int JSONB_TYPE_STRING = 0xC;
public static final int JSONB_TYPE_OPAQUE = 0xF;
public static final char JSONB_NULL_LITERAL = '\0';
public static final char JSONB_TRUE_LITERAL = '\1';
public static final char JSONB_FALSE_LITERAL = '\2';
/*
* The size of offset or size fields in the small and the large storage
* format for JSON objects and JSON arrays.
*/
public static final int SMALL_OFFSET_SIZE = 2;
public static final int LARGE_OFFSET_SIZE = 4;
/*
* The size of key entries for objects when using the small storage format
* or the large storage format. In the small format it is 4 bytes (2 bytes
* for key length and 2 bytes for key offset). In the large format it is 6
* (2 bytes for length, 4 bytes for offset).
*/
public static final int KEY_ENTRY_SIZE_SMALL = (2 + SMALL_OFFSET_SIZE);
public static final int KEY_ENTRY_SIZE_LARGE = (2 + LARGE_OFFSET_SIZE);
/*
* The size of value entries for objects or arrays. When using the small
* storage format, the entry size is 3 (1 byte for type, 2 bytes for
* offset). When using the large storage format, it is 5 (1 byte for type, 4
* bytes for offset).
*/
public static final int VALUE_ENTRY_SIZE_SMALL = (1 + SMALL_OFFSET_SIZE);
public static final int VALUE_ENTRY_SIZE_LARGE = (1 + LARGE_OFFSET_SIZE);
public static Json_Value parse_value(int type, LogBuffer buffer, long len) {
buffer = buffer.duplicate(buffer.position(), (int) len);
switch (type) {
case JSONB_TYPE_SMALL_OBJECT:
return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false);
case JSONB_TYPE_LARGE_OBJECT:
return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true);
case JSONB_TYPE_SMALL_ARRAY:
return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false);
case JSONB_TYPE_LARGE_ARRAY:
return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true);
default:
return parse_scalar(type, buffer, len);
}
}
private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large) {
long offset_size = large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
if (len < 2 * offset_size) {
throw new IllegalArgumentException("illegal json data");
}
long element_count = read_offset_or_size(buffer, large);
long bytes = read_offset_or_size(buffer, large);
if (bytes > len) {
throw new IllegalArgumentException("illegal json data");
}
long header_size = 2 * offset_size;
if (type == Json_enum_type.OBJECT) {
header_size += element_count * (large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL);
}
header_size += element_count * (large ? VALUE_ENTRY_SIZE_LARGE : VALUE_ENTRY_SIZE_SMALL);
if (header_size > bytes) {
throw new IllegalArgumentException("illegal json data");
}
return new Json_Value(type, buffer.rewind(), element_count, bytes, large);
}
private static long read_offset_or_size(LogBuffer buffer, boolean large) {
return large ? buffer.getUint32() : buffer.getUint16();
}
private static Json_Value parse_scalar(int type, LogBuffer buffer, long len) {
switch (type) {
case JSONB_TYPE_LITERAL:
/* purecov: inspected */
int data = buffer.getUint8();
switch (data) {
case JSONB_NULL_LITERAL:
return new Json_Value(Json_enum_type.LITERAL_NULL);
case JSONB_TRUE_LITERAL:
return new Json_Value(Json_enum_type.LITERAL_TRUE);
case JSONB_FALSE_LITERAL:
return new Json_Value(Json_enum_type.LITERAL_FALSE);
default:
throw new IllegalArgumentException("illegal json data");
}
case JSONB_TYPE_INT16:
return new Json_Value(Json_enum_type.INT, buffer.getInt16());
case JSONB_TYPE_INT32:
return new Json_Value(Json_enum_type.INT, buffer.getInt32());
case JSONB_TYPE_INT64:
return new Json_Value(Json_enum_type.INT, buffer.getLong64());
case JSONB_TYPE_UINT16:
return new Json_Value(Json_enum_type.UINT, buffer.getUint16());
case JSONB_TYPE_UINT32:
return new Json_Value(Json_enum_type.UINT, buffer.getUint32());
case JSONB_TYPE_UINT64:
return new Json_Value(Json_enum_type.UINT, buffer.getUlong64());
case JSONB_TYPE_DOUBLE:
return new Json_Value(Json_enum_type.DOUBLE, Double.valueOf(buffer.getDouble64()));
case JSONB_TYPE_STRING:
int max_bytes = (int) Math.min(len, 5);
long tlen = 0;
long str_len = 0;
long n = 0;
byte[] datas = buffer.getData(max_bytes);
for (int i = 0; i < max_bytes; i++) {
// Get the next 7 bits of the length.
tlen |= (datas[i] & 0x7f) << (7 * i);
if ((datas[i] & 0x80) == 0) {
// The length shouldn't exceed 32 bits.
if (tlen > 4294967296L) {
throw new IllegalArgumentException("illegal json data");
}
// This was the last byte. Return successfully.
n = i + 1;
str_len = tlen;
break;
}
}
if (str_len == 0 || len < n + str_len) {
throw new IllegalArgumentException("illegal json data");
}
return new Json_Value(Json_enum_type.STRING, buffer.rewind()
.forward((int) n)
.getFixString((int) str_len));
case JSONB_TYPE_OPAQUE:
/*
* There should always be at least one byte, which tells the
* field type of the opaque value.
*/
// The type is encoded as a uint8 that maps to an
// enum_field_types.
int type_byte = buffer.getUint8();
int position = buffer.position();
// Then there's the length of the value.
int q_max_bytes = (int) Math.min(len, 5);
long q_tlen = 0;
long q_str_len = 0;
long q_n = 0;
byte[] q_datas = buffer.getData(q_max_bytes);
for (int i = 0; i < q_max_bytes; i++) {
// Get the next 7 bits of the length.
q_tlen |= (q_datas[i] & 0x7f) << (7 * i);
if ((q_datas[i] & 0x80) == 0) {
// The length shouldn't exceed 32 bits.
if (q_tlen > 4294967296L) {
throw new IllegalArgumentException("illegal json data");
}
// This was the last byte. Return successfully.
q_n = i + 1;
q_str_len = q_tlen;
break;
}
}
if (q_str_len == 0 || len < q_n + q_str_len) {
throw new IllegalArgumentException("illegal json data");
}
return new Json_Value(type_byte, buffer.position(position).forward((int) q_n), q_str_len);
default:
throw new IllegalArgumentException("illegal json data");
}
}
public static class Json_Value {
Json_enum_type m_type;
int m_field_type;
LogBuffer m_data;
long m_element_count;
long m_length;
String m_string_value;
Number m_int_value;
double m_double_value;
boolean m_large;
public Json_Value(Json_enum_type t){
this.m_type = t;
}
public Json_Value(Json_enum_type t, Number val){
this.m_type = t;
if (t == Json_enum_type.DOUBLE) {
this.m_double_value = val.doubleValue();
} else {
this.m_int_value = val;
}
}
public Json_Value(Json_enum_type t, String value){
this.m_type = t;
this.m_string_value = value;
}
public Json_Value(int field_type, LogBuffer data, long bytes){
this.m_type = Json_enum_type.OPAQUE; // 不确定类型
this.m_field_type = field_type;
this.m_data = data;
this.m_length = bytes;
}
public Json_Value(Json_enum_type t, LogBuffer data, long element_count, long bytes, boolean large){
this.m_type = t;
this.m_data = data;
this.m_element_count = element_count;
this.m_length = bytes;
this.m_large = large;
}
public String key(int i) {
m_data.rewind();
int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
int entry_offset = 2 * offset_size + key_entry_size * i;
// The offset of the key is the first part of the key
// entry.
m_data.forward(entry_offset);
long key_offset = read_offset_or_size(m_data, m_large);
// The length of the key is the second part of the
// entry, always two
// bytes.
long key_length = m_data.getUint16();
return m_data.rewind().forward((int) key_offset).getFixString((int) key_length);
}
public Json_Value element(int i) {
m_data.rewind();
int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
int value_entry_size = m_large ? VALUE_ENTRY_SIZE_LARGE : VALUE_ENTRY_SIZE_SMALL;
int first_entry_offset = 2 * offset_size;
if (m_type == Json_enum_type.OBJECT) {
first_entry_offset += m_element_count * key_entry_size;
}
int entry_offset = first_entry_offset + value_entry_size * i;
int type = m_data.forward(entry_offset).getUint8();
if (type == JSONB_TYPE_INT16 || type == JSONB_TYPE_UINT16 || type == JSONB_TYPE_LITERAL
|| (m_large && (type == JSONB_TYPE_INT32 || type == JSONB_TYPE_UINT32))) {
return parse_scalar(type, m_data, value_entry_size - 1);
}
int value_offset = (int) read_offset_or_size(m_data, m_large);
return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset);
}
public StringBuilder toJsonString(StringBuilder buf) {
switch (m_type) {
case OBJECT:
buf.append("{");
for (int i = 0; i < m_element_count; ++i) {
if (i > 0) {
buf.append(", ");
}
buf.append('"').append(key(i)).append('"');
buf.append(": ");
element(i).toJsonString(buf);
}
buf.append("}");
break;
case ARRAY:
buf.append("[");
for (int i = 0; i < m_element_count; ++i) {
if (i > 0) {
buf.append(", ");
}
element(i).toJsonString(buf);
}
buf.append("]");
break;
case DOUBLE:
buf.append(Double.valueOf(m_double_value).toString());
break;
case INT:
buf.append(m_int_value.toString());
break;
case UINT:
buf.append(m_int_value.toString());
break;
case LITERAL_FALSE:
buf.append("false");
break;
case LITERAL_TRUE:
buf.append("true");
break;
case LITERAL_NULL:
buf.append("NULL");
break;
case OPAQUE:
String text = null;
if (m_field_type == LogEvent.MYSQL_TYPE_NEWDECIMAL) {
int precision = m_data.getInt8();
int scale = m_data.getInt8();
text = m_data.getDecimal(precision, scale).toPlainString();
buf.append(text);
} else if (m_field_type == LogEvent.MYSQL_TYPE_TIME) {
long packed_value = m_data.getLong64();
if (packed_value == 0) {
text = "00:00:00";
} else {
long ultime = Math.abs(packed_value);
long intpart = ultime >> 24;
int frac = (int) (ultime % (1L << 24));
text = String.format("%s%02d:%02d:%02d",
packed_value >= 0 ? "" : "-",
(int) ((intpart >> 12) % (1 << 10)),
(int) ((intpart >> 6) % (1 << 6)),
(int) (intpart % (1 << 6)));
text = text + "." + usecondsToStr(frac, 6);
}
buf.append('"').append(text).append('"');
} else if (m_field_type == LogEvent.MYSQL_TYPE_DATE || m_field_type == LogEvent.MYSQL_TYPE_DATETIME
|| m_field_type == LogEvent.MYSQL_TYPE_TIMESTAMP) {
long packed_value = m_data.getLong64();
if (packed_value == 0) {
text = "0000-00-00 00:00:00";
} else {
// 构造TimeStamp只处理到秒
long ultime = Math.abs(packed_value);
long intpart = ultime >> 24;
int frac = (int) (ultime % (1L << 24));
long ymd = intpart >> 17;
long ym = ymd >> 5;
long hms = intpart % (1 << 17);
text = String.format("%04d-%02d-%02d %02d:%02d:%02d",
(int) (ym / 13),
(int) (ym % 13),
(int) (ymd % (1 << 5)),
(int) (hms >> 12),
(int) ((hms >> 6) % (1 << 6)),
(int) (hms % (1 << 6)));
text = text + "." + usecondsToStr(frac, 6);
}
buf.append('"').append(text).append('"');
} else {
text = m_data.getFixString((int) m_length);
buf.append('"').append(text).append('"');
}
break;
case STRING:
buf.append('"').append(m_string_value).append('"');
break;
case ERROR:
throw new IllegalArgumentException("illegal json data");
}
return buf;
}
}
public static enum Json_enum_type {
OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR
}
private static String usecondsToStr(int frac, int meta) {
String sec = String.valueOf(frac);
if (meta > 6) {
throw new IllegalArgumentException("unknow useconds meta : " + meta);
}
if (sec.length() < 6) {
StringBuilder result = new StringBuilder(6);
int len = 6 - sec.length();
for (; len > 0; len--) {
result.append('0');
}
result.append(sec);
sec = result.toString();
}
return sec.substring(0, meta);
}
}

View File

@@ -267,6 +267,7 @@ public abstract class LogEvent {
public static final int MYSQL_TYPE_TIMESTAMP2 = 17;
public static final int MYSQL_TYPE_DATETIME2 = 18;
public static final int MYSQL_TYPE_TIME2 = 19;
public static final int MYSQL_TYPE_JSON = 245;
public static final int MYSQL_TYPE_NEWDECIMAL = 246;
public static final int MYSQL_TYPE_ENUM = 247;
public static final int MYSQL_TYPE_SET = 248;

View File

@@ -140,10 +140,15 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
throw new IOException("Unknown binlog version: " + binlogVersion);
}
}
public FormatDescriptionLogEvent(final int binlogVersion,int binlogChecksum){
this(binlogVersion);
this.header.checksumAlg = binlogChecksum;
}
public FormatDescriptionLogEvent(final int binlogVersion){
this.binlogVersion = binlogVersion;
postHeaderLen = new short[ENUM_END_EVENT];
/* identify binlog format */
switch (binlogVersion) {

View File

@@ -90,6 +90,7 @@ public final class RotateLogEvent extends LogEvent {
int filenameLen = buffer.limit() - filenameOffset;
if (filenameLen > FN_REFLEN - 1) filenameLen = FN_REFLEN - 1;
buffer.position(filenameOffset);
filename = buffer.getFixString(filenameLen);
}

View File

@@ -8,6 +8,8 @@ import java.util.BitSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.taobao.tddl.dbsync.binlog.JsonConversion;
import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -25,6 +27,7 @@ public final class RowsLogBuffer {
public static final long DATETIMEF_INT_OFS = 0x8000000000L;
public static final long TIMEF_INT_OFS = 0x800000L;
public static final long TIMEF_OFS = 0x800000000000L;
private final LogBuffer buffer;
private final int columnLen;
private final String charsetName;
@@ -163,10 +166,13 @@ public final class RowsLogBuffer {
case LogEvent.MYSQL_TYPE_TIMESTAMP:
case LogEvent.MYSQL_TYPE_DATETIME:
case LogEvent.MYSQL_TYPE_TIMESTAMP2:
case LogEvent.MYSQL_TYPE_DATETIME2:
javaType = Types.TIMESTAMP;
break;
case LogEvent.MYSQL_TYPE_TIME:
case LogEvent.MYSQL_TYPE_TIME2:
javaType = Types.TIME;
break;
@@ -942,6 +948,22 @@ public final class RowsLogBuffer {
length = len;
break;
}
case LogEvent.MYSQL_TYPE_JSON: {
len = buffer.getUint16();
buffer.forward(meta - 2);
int position = buffer.position();
Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1);
StringBuilder builder = new StringBuilder();
jsonValue.toJsonString(builder);
value = builder.toString();
buffer.position(position + len);
// byte[] binary = new byte[len];
// buffer.fillBytes(binary, 0, len);
// value = binary;
javaType = Types.VARCHAR;
length = len;
break;
}
case LogEvent.MYSQL_TYPE_GEOMETRY: {
/*
* MYSQL_TYPE_GEOMETRY: copy from BLOB or TEXT
@@ -1023,4 +1045,5 @@ public final class RowsLogBuffer {
return sec.substring(0, meta);
}
}

View File

@@ -402,6 +402,7 @@ public final class TableMapLogEvent extends LogEvent {
case MYSQL_TYPE_DOUBLE:
case MYSQL_TYPE_FLOAT:
case MYSQL_TYPE_GEOMETRY:
case MYSQL_TYPE_JSON:
/*
* These types store a single byte.
*/

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -5,6 +5,7 @@ import java.util.Properties;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.Watcher.Event.KeeperState;
@@ -88,9 +89,9 @@ public class CanalController {
cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
ip = getProperty(properties, CanalConstants.CANAL_IP);
port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
embededCanalServer = new CanalServerWithEmbedded();
embededCanalServer = CanalServerWithEmbedded.instance();
embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
canalServer = new CanalServerWithNetty(embededCanalServer);
canalServer = CanalServerWithNetty.instance();
canalServer.setIp(ip);
canalServer.setPort(port);
@@ -172,6 +173,8 @@ public class CanalController {
if (zkclientx != null) {
runningMonitor.setZkClient(zkclientx);
}
// 触发创建一下cid节点
runningMonitor.init();
return runningMonitor;
}
}));
@@ -189,10 +192,10 @@ public class CanalController {
instanceConfigs.put(destination, config);
}
if (!config.getLazy() && !embededCanalServer.isStart(destination)) {
if (!embededCanalServer.isStart(destination)) {
// HA机制启动
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
if (!runningMonitor.isStart()) {
if (!config.getLazy() && !runningMonitor.isStart()) {
runningMonitor.start();
}
}
@@ -231,7 +234,13 @@ public class CanalController {
if (StringUtils.isEmpty(rootDir)) {
rootDir = "../conf";
}
monitor.setRootConf(rootDir);
if (StringUtils.equals("otter-canal", System.getProperty("appName"))) {
monitor.setRootConf(rootDir);
} else {
// eclipse debug模式
monitor.setRootConf("src/main/resources/");
}
return monitor;
} else if (mode.isManager()) {
return new ManagerInstanceConfigMonitor();
@@ -383,10 +392,10 @@ public class CanalController {
final String destination = entry.getKey();
InstanceConfig config = entry.getValue();
// 创建destination的工作节点
if (!config.getLazy() && !embededCanalServer.isStart(destination)) {
if (!embededCanalServer.isStart(destination)) {
// HA机制启动
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
if (!runningMonitor.isStart()) {
if (!config.getLazy() && !runningMonitor.isStart()) {
runningMonitor.start();
}
}
@@ -442,6 +451,9 @@ public class CanalController {
String parentDir = path.substring(0, path.lastIndexOf('/'));
zkclientx.createPersistent(parentDir, true);
zkclientx.createEphemeral(path);
} catch (ZkNodeExistsException e) {
// ignore
// 因为第一次启动时创建了cid,但在stop/start的时可能会关闭和新建,允许出现NodeExists问题s
}
}

View File

@@ -175,6 +175,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
InstanceAction action = actions.remove(destination);
try {
action.stop(destination);
lastFiles.remove(destination);
logger.info("auto notify stop {} successful.", destination);
} catch (Throwable e) {
logger.error("scan delete found[{}] but stop failed", destination, ExceptionUtils.getFullStackTrace(e));

View File

@@ -40,6 +40,7 @@ canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED
@@ -63,4 +64,4 @@ canal.instance.global.lazy = false
#canal.instance.global.manager.address = 127.0.0.1:1099
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

View File

@@ -178,8 +178,9 @@
<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
<property name="filterRows" value="${canal.instance.filter.rows:false}" />
<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
</bean>
</beans>
</beans>

View File

@@ -163,8 +163,9 @@
<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
<property name="filterRows" value="${canal.instance.filter.rows:false}" />
<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
</bean>
</beans>
</beans>

View File

@@ -250,8 +250,9 @@
<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
<property name="filterRows" value="${canal.instance.filter.rows:false}" />
<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
</bean>
</beans>
</beans>

View File

@@ -0,0 +1,163 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:lang="http://www.springframework.org/schema/lang"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-2.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"
default-autowire="byName">
<!-- properties -->
<bean class="com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer" lazy-init="false">
<property name="ignoreResourceNotFound" value="true" />
<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/><!-- 允许system覆盖 -->
<property name="locationNames">
<list>
<value>classpath:canal.properties</value>
<value>classpath:${canal.instance.destination:}/instance.properties</value>
</list>
</property>
</bean>
<bean id="socketAddressEditor" class="com.alibaba.otter.canal.instance.spring.support.SocketAddressEditor" />
<bean class="org.springframework.beans.factory.config.CustomEditorConfigurer">
<property name="propertyEditorRegistrars">
<list>
<ref bean="socketAddressEditor" />
</list>
</property>
</bean>
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" />
<property name="eventParser">
<ref local="eventParser" />
</property>
<property name="eventSink">
<ref local="eventSink" />
</property>
<property name="eventStore">
<ref local="eventStore" />
</property>
<property name="metaManager">
<ref local="metaManager" />
</property>
<property name="alarmHandler">
<ref local="alarmHandler" />
</property>
<property name="eventFetcher">
<ref local="eventFetcher"/>
</property>
</bean>
<bean id="eventFetcher" class="com.alibaba.otter.canal.mq.fetch.CanalEventFetcherWithMq">
<property name="enable" value="${canal.instance.mq.enable}"/>
<property name="sendMode" value="${canal.instance.mq.sendMode}"/>
</bean>
<!-- 报警处理类 -->
<bean id="alarmHandler" class="com.alibaba.otter.canal.common.alarm.MsgAlarmHandler">
<property name="enabled" value="${canal.alarm.enable}" />
<property name="environment" value="${canal.alarm.environment}" />
<property name="intercenterApi" value="${canal.alarm.intercenter.api}" />
<property name="administrators" value="${canal.alarm.administrators}" />
</bean>
<bean id="zkClientx" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean" >
<property name="targetClass" value="com.alibaba.otter.canal.common.zookeeper.ZkClientx" />
<property name="targetMethod" value="getZkClient" />
<property name="arguments">
<list>
<value>${canal.zkServers:127.0.0.1:2181}</value>
</list>
</property>
</bean>
<bean id="metaManager" class="com.alibaba.otter.canal.meta.PeriodMixedMetaManager">
<property name="zooKeeperMetaManager">
<bean class="com.alibaba.otter.canal.meta.ZooKeeperMetaManager">
<property name="zkClientx" ref="zkClientx" />
</bean>
</property>
<property name="period" value="${canal.zookeeper.flush.period:1000}" />
</bean>
<bean id="eventStore" class="com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer">
<property name="bufferSize" value="${canal.instance.memory.buffer.size:16384}" />
<property name="bufferMemUnit" value="${canal.instance.memory.buffer.memunit:1024}" />
<property name="batchMode" value="${canal.instance.memory.batch.mode:MEMSIZE}" />
<property name="ddlIsolation" value="${canal.instance.get.ddl.isolation:false}" />
</bean>
<bean id="eventSink" class="com.alibaba.otter.canal.sink.entry.EntryEventSink">
<property name="eventStore" ref="eventStore" />
</bean>
<bean id="eventParser" class="com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser">
<property name="destination" value="${canal.instance.destination}" />
<property name="alarmHandler" ref="alarmHandler" />
<!-- 解析过滤处理 -->
<property name="eventFilter">
<bean class="com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter" >
<constructor-arg index="0" value="${canal.instance.filter.regex:.*\..*}" />
</bean>
</property>
<property name="eventBlackFilter">
<bean class="com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter" >
<constructor-arg index="0" value="${canal.instance.filter.black.regex:}" />
<constructor-arg index="1" value="false" />
</bean>
</property>
<!-- 最大事务解析大小,超过该大小后事务将被切分为多个事务投递 -->
<property name="transactionSize" value="${canal.instance.transaction.size:1024}" />
<!-- 解析编码 -->
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
<!-- 解析位点记录 -->
<property name="logPositionManager">
<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
<property name="primary">
<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
</property>
<property name="failback">
<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
<property name="metaManager" ref="metaManager" />
</bean>
</property>
</bean>
</property>
<!-- 解析数据库信息 -->
<property name="masterInfo">
<bean class="com.alibaba.otter.canal.parse.support.AuthenticationInfo">
<property name="address" value="${canal.instance.master.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
</bean>
</property>
<!-- 解析起始位点 -->
<property name="masterPosition">
<bean class="com.alibaba.otter.canal.protocol.position.EntryPosition">
<property name="journalName" value="${canal.instance.master.journal.name}" />
<property name="position" value="${canal.instance.master.position}" />
<property name="timestamp" value="${canal.instance.master.timestamp}" />
</bean>
</property>
<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
<property name="filterRows" value="${canal.instance.filter.rows:false}" />
<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
<property name="needWait" value="${canal.instance.parser.needWait:true}"/>
<property name="directory" value="${canal.instance.parser.directory}"/>
</bean>
</beans>

View File

@@ -151,8 +151,9 @@
<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
<property name="filterRows" value="${canal.instance.filter.rows:false}" />
<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
</bean>
</beans>
</beans>

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -12,6 +12,7 @@ import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.HandshakeInitializationPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.Reply323Packet;
@@ -26,25 +27,27 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
*/
public class MysqlConnector {
private static final Logger logger = LoggerFactory.getLogger(MysqlConnector.class);
private InetSocketAddress address;
private String username;
private String password;
private static final Logger logger = LoggerFactory.getLogger(MysqlConnector.class);
private InetSocketAddress address;
private String username;
private String password;
private byte charsetNumber = 33;
private String defaultSchema = "retl";
private int soTimeout = 30 * 1000;
private int receiveBufferSize = 16 * 1024;
private int sendBufferSize = 16 * 1024;
private byte charsetNumber = 33;
private String defaultSchema = "retl";
private int soTimeout = 30 * 1000;
private int receiveBufferSize = 16 * 1024;
private int sendBufferSize = 16 * 1024;
private SocketChannel channel;
private AtomicBoolean connected = new AtomicBoolean(false);
private SocketChannel channel;
private volatile boolean dumping = false;
// mysql connectinnId
private long connectionId = -1;
private AtomicBoolean connected = new AtomicBoolean(false);
public MysqlConnector(){
}
public MysqlConnector(InetSocketAddress address, String username, String password){
this.address = address;
this.username = username;
this.password = password;
@@ -86,11 +89,30 @@ public class MysqlConnector {
if (channel != null) {
channel.close();
}
logger.info("disConnect MysqlConnection to {}...", address);
} catch (Exception e) {
throw new IOException("disconnect " + this.address + " failure:" + ExceptionUtils.getStackTrace(e));
}
// 执行一次quit
if (dumping && connectionId >= 0) {
MysqlConnector connector = null;
try {
connector = this.fork();
connector.connect();
MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector);
executor.update("KILL CONNECTION " + connectionId);
} catch (Exception e) {
// 忽略具体异常
logger.info("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e));
} finally {
if (connector != null) {
connector.disconnect();
}
}
dumping = false;
}
} else {
logger.info("the channel {} is not connected", this.address);
}
@@ -113,6 +135,17 @@ public class MysqlConnector {
return connector;
}
public void quit() throws IOException {
QuitCommandPacket quit = new QuitCommandPacket();
byte[] cmdBody = quit.toBytes();
HeaderPacket quitHeader = new HeaderPacket();
quitHeader.setPacketBodyLength(cmdBody.length);
quitHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.write(channel,
new ByteBuffer[] { ByteBuffer.wrap(quitHeader.toBytes()), ByteBuffer.wrap(cmdBody) });
}
// ====================== help method ====================
private void configChannel(SocketChannel channel) throws IOException {
@@ -140,6 +173,7 @@ public class MysqlConnector {
}
HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
handshakePacket.fromBytes(body);
connectionId = handshakePacket.threadId; // 记录一下connection
logger.info("handshake initialization packet received, prepare the client authentication packet to send");
@@ -291,4 +325,20 @@ public class MysqlConnector {
this.password = password;
}
public long getConnectionId() {
return connectionId;
}
public void setConnectionId(long connectionId) {
this.connectionId = connectionId;
}
public boolean isDumping() {
return dumping;
}
public void setDumping(boolean dumping) {
this.dumping = dumping;
}
}

View File

@@ -0,0 +1,35 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
/**
* quit cmd
*
* @author agapple 2016年3月1日 下午8:33:02
* @since 1.0.22
*/
public class QuitCommandPacket extends CommandPacket {
public static final byte[] QUIT = new byte[] { 1, 0, 0, 0, 1 };
public QuitCommandPacket(){
setCommand((byte) 0x01);
}
@Override
public void fromBytes(byte[] data) throws IOException {
}
@Override
public byte[] toBytes() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
out.write(getCommand());
out.write(QUIT);
return out.toByteArray();
}
}

View File

@@ -74,6 +74,8 @@ public class FieldPacket extends PacketWithHeaderPacket {
this.decimals = data[index];
index++;
//
index += 2;//skip filter
//
if (index < data.length) {
reader.setIndex(index);
this.definition = reader.readLengthCodedString(data);

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

90
example/src/main/bin/init.sh Executable file
View File

@@ -0,0 +1,90 @@
#! /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}/..
# default values
PORT=3306
MYSQL_IP="0"
# parse arguments
while getopts "h:p:" opt; do
case $opt in
p)
if [[ $OPTARG -gt 0 ]]; then
PORT=$OPTARG
else
echo "Invalid -p option: -$OPTARG, should be positive integer >&2"
exit 1
fi
;;
h)
MYSQL_IP=$OPTARG
;;
\?)
printUsage
exit 1
;;
esac
done
function printUsage
{
cat << __Usage
Initialise the instance config property file automatically. This tool will connect to Mysql master server and get the binlog file and position then update the instance config property file for you.
Usage: $0 -p [port] -h [hostname]
Options:
-p: mysql port, default to 3306
-h: mysql host ip.
For example:
$0 -p 3306 -h 172.17.0.2
Enjoy!!!
__Usage
}
if [[ $PORT -lt 0 ]]; then
echo "invalid port argument $PORT"
printUsage
exit 1
fi
# check if mysql ip is specified
if [[ $MYSQL_IP = "0" ]]; then
echo '-h mysql host ip argument is required.'
printUsage
exit 1
fi
MYSQL_ADDR="$MYSQL_IP:$PORT"
MASTER_STATUS=`mysql -ucanal -pcanal -h$MYSQL_IP -P$PORT -e 'show master status' 2>/dev/null | head -n 2 | tail -n 1`
LOG_FILE=`echo "$MASTER_STATUS" | cut -f 1`
LOG_POS=`echo "$MASTER_STATUS" | cut -f 2`
INSTANCE_CONF_FILE=$base/conf/example/instance.properties
if [ ! -f $INSTANCE_CONF_FILE ]; then
echo "Can not find instance config file: $INSTANCE_CONF_FILE"
exit 1
fi
# replace the binlog file and position
sed -i.bak -e "s/canal\.instance\.master\.address = .*/canal\.instance\.master\.address = $MYSQL_ADDR/" -e "s/canal\.instance\.master\.journal\.name = .*/canal\.instance\.master\.journal\.name = $LOG_FILE/" -e "s/canal\.instance\.master\.position = .*/canal\.instance\.master\.position = $LOG_POS/" $INSTANCE_CONF_FILE
cat $INSTANCE_CONF_FILE
echo
echo 'Above is your instance config file, check if the settings are correct'
echo

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<artifactId>canal.instance.core</artifactId>

View File

@@ -0,0 +1,247 @@
package com.alibaba.otter.canal.instance.core;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.ha.CanalHAController;
import com.alibaba.otter.canal.parse.ha.HeartBeatHAController;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
/**
* Created with Intellig IDEA.
* Author: yinxiu
* Date: 2016-01-07
* Time: 22:26
*/
public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
protected Long canalId; // 和manager交互唯一标示
protected String destination; // 队列名字
protected CanalEventStore<Event> eventStore; // 有序队列
protected CanalEventParser eventParser; // 解析对应的数据信息
protected CanalEventSink<List<CanalEntry.Entry>> eventSink; // 链接parse和store的桥接器
protected CanalMetaManager metaManager; // 消费信息管理器
protected CanalAlarmHandler alarmHandler; // alarm报警机制
@Override
public boolean subscribeChange(ClientIdentity identity) {
if (StringUtils.isNotEmpty(identity.getFilter())) {
logger.info("subscribe filter change to " + identity.getFilter());
AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
}
} else {
((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
}
}
// filter的处理规则
// a. parser处理数据过滤处理
// b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份,每份的数据可以根据自己的过滤规则不同而有不同的数据
// 后续内存版的一对多分发,可以考虑
return true;
}
@Override
public void start() {
super.start();
if (!metaManager.isStart()) {
metaManager.start();
}
if (!alarmHandler.isStart()) {
alarmHandler.start();
}
if (!eventStore.isStart()) {
eventStore.start();
}
if (!eventSink.isStart()) {
eventSink.start();
}
if (!eventParser.isStart()) {
beforeStartEventParser(eventParser);
eventParser.start();
afterStartEventParser(eventParser);
}
logger.info("start successful....");
}
@Override
public void stop() {
super.stop();
logger.info("stop CannalInstance for {}-{} ", new Object[] { canalId, destination });
if (eventParser.isStart()) {
beforeStopEventParser(eventParser);
eventParser.stop();
afterStopEventParser(eventParser);
}
if (eventSink.isStart()) {
eventSink.stop();
}
if (eventStore.isStart()) {
eventStore.stop();
}
if (metaManager.isStart()) {
metaManager.stop();
}
if (alarmHandler.isStart()) {
alarmHandler.stop();
}
logger.info("stop successful....");
}
protected void beforeStartEventParser(CanalEventParser eventParser) {
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
startEventParserInternal(singleEventParser, true);
}
} else {
startEventParserInternal(eventParser, false);
}
}
// around event parser, default impl
protected void afterStartEventParser(CanalEventParser eventParser) {
// 读取一下历史订阅的filter信息
List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
for (ClientIdentity clientIdentity : clientIdentitys) {
subscribeChange(clientIdentity);
}
}
// around event parser
protected void beforeStopEventParser(CanalEventParser eventParser) {
// noop
}
protected void afterStopEventParser(CanalEventParser eventParser) {
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
stopEventParserInternal(singleEventParser);
}
} else {
stopEventParserInternal(eventParser);
}
}
/**
* 初始化单个eventParser,不需要考虑group
*/
protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
if (eventParser instanceof AbstractEventParser) {
AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
// 首先启动log position管理器
CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
if (!logPositionManager.isStart()) {
logPositionManager.start();
}
}
if (eventParser instanceof MysqlEventParser) {
MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
CanalHAController haController = mysqlEventParser.getHaController();
if (haController instanceof HeartBeatHAController) {
((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
}
if (!haController.isStart()) {
haController.start();
}
}
}
protected void stopEventParserInternal(CanalEventParser eventParser) {
if (eventParser instanceof AbstractEventParser) {
AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
// 首先启动log position管理器
CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
if (logPositionManager.isStart()) {
logPositionManager.stop();
}
}
if (eventParser instanceof MysqlEventParser) {
MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
CanalHAController haController = mysqlEventParser.getHaController();
if (haController.isStart()) {
haController.stop();
}
}
}
// ==================getter==================================
@Override
public String getDestination() {
return destination;
}
@Override
public CanalEventParser getEventParser() {
return eventParser;
}
@Override
public CanalEventSink getEventSink() {
return eventSink;
}
@Override
public CanalEventStore getEventStore() {
return eventStore;
}
@Override
public CanalMetaManager getMetaManager() {
return metaManager;
}
@Override
public CanalAlarmHandler getAlarmHandler() {
return alarmHandler;
}
}

View File

@@ -1,105 +0,0 @@
package com.alibaba.otter.canal.instance.core;
import java.util.List;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.ha.CanalHAController;
import com.alibaba.otter.canal.parse.ha.HeartBeatHAController;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
/**
* @author zebin.xuzb 2012-10-17 下午3:12:34
* @version 1.0.0
*/
public abstract class CanalInstanceSupport extends AbstractCanalLifeCycle {
protected void beforeStartEventParser(CanalEventParser eventParser) {
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
startEventParserInternal(singleEventParser, true);
}
} else {
startEventParserInternal(eventParser, false);
}
}
// around event parser
protected void afterStartEventParser(CanalEventParser eventParser) {
// noop
}
// around event parser
protected void beforeStopEventParser(CanalEventParser eventParser) {
// noop
}
protected void afterStopEventParser(CanalEventParser eventParser) {
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
stopEventParserInternal(singleEventParser);
}
} else {
stopEventParserInternal(eventParser);
}
}
/**
* 初始化单个eventParser,不需要考虑group
*/
protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
if (eventParser instanceof AbstractEventParser) {
AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
// 首先启动log position管理器
CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
if (!logPositionManager.isStart()) {
logPositionManager.start();
}
}
if (eventParser instanceof MysqlEventParser) {
MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
CanalHAController haController = mysqlEventParser.getHaController();
if (haController instanceof HeartBeatHAController) {
((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
}
if (!haController.isStart()) {
haController.start();
}
}
}
protected void stopEventParserInternal(CanalEventParser eventParser) {
if (eventParser instanceof AbstractEventParser) {
AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
// 首先启动log position管理器
CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
if (logPositionManager.isStart()) {
logPositionManager.stop();
}
}
if (eventParser instanceof MysqlEventParser) {
MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
CanalHAController haController = mysqlEventParser.getHaController();
if (haController.isStart()) {
haController.stop();
}
}
}
}

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -17,8 +17,7 @@ import com.alibaba.otter.canal.common.alarm.LogAlarmHandler;
import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceSupport;
import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.DataSourcing;
@@ -28,7 +27,6 @@ import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageScavengeMode;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.meta.MemoryMetaManager;
import com.alibaba.otter.canal.meta.PeriodMixedMetaManager;
import com.alibaba.otter.canal.meta.ZooKeeperMetaManager;
@@ -46,17 +44,12 @@ import com.alibaba.otter.canal.parse.index.MetaLogPositionManager;
import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager;
import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.sink.entry.EntryEventSink;
import com.alibaba.otter.canal.sink.entry.group.GroupEventSink;
import com.alibaba.otter.canal.store.AbstractCanalStoreScavenge;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.BatchMode;
import com.alibaba.otter.canal.store.model.Event;
/**
* 单个canal实例,比如一个destination会独立一个实例
@@ -64,24 +57,11 @@ import com.alibaba.otter.canal.store.model.Event;
* @author jianghang 2012-7-11 下午09:26:51
* @version 1.0.0
*/
public class CanalInstanceWithManager extends CanalInstanceSupport implements CanalInstance {
public class CanalInstanceWithManager extends AbstractCanalInstance {
private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithManager.class);
protected Long canalId; // 和manager交互唯一标示
protected String destination; // 队列名字
protected String filter; // 过滤表达式
protected CanalParameter parameters; // 对应参数
protected CanalMetaManager metaManager; // 消费信息管理器
protected CanalEventStore<Event> eventStore; // 有序队列
protected CanalEventParser eventParser; // 解析对应的数据信息
protected CanalEventSink<List<Entry>> eventSink; // 链接parse和store的桥接器
protected CanalAlarmHandler alarmHandler; // alarm报警机制
protected ZkClientx zkClientx;
public CanalInstanceWithManager(Canal canal){
this(canal, null);
}
private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithManager.class);
protected String filter; // 过滤表达式
protected CanalParameter parameters; // 对应参数
public CanalInstanceWithManager(Canal canal, String filter){
this.parameters = canal.getCanalParameter();
@@ -89,7 +69,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
this.destination = canal.getName();
this.filter = filter;
logger.info("init CannalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
logger.info("init CanalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
// 初始化报警机制
initAlarmHandler();
// 初始化metaManager
@@ -101,7 +81,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
// 初始化eventParser;
initEventParser();
// 基础工具,需要提前start,会有先订阅再根据filter条件启动paser的需求
// 基础工具,需要提前start,会有先订阅再根据filter条件启动parse的需求
if (!alarmHandler.isStart()) {
alarmHandler.start();
}
@@ -113,98 +93,9 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
}
public void start() {
super.start();
// 初始化metaManager
logger.info("start CannalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
if (!metaManager.isStart()) {
metaManager.start();
}
if (!alarmHandler.isStart()) {
alarmHandler.start();
}
if (!eventStore.isStart()) {
eventStore.start();
}
if (!eventSink.isStart()) {
eventSink.start();
}
if (!eventParser.isStart()) {
beforeStartEventParser(eventParser);
eventParser.start();
}
logger.info("start successful....");
}
public void stop() {
logger.info("stop CannalInstance for {}-{} ", new Object[] { canalId, destination });
if (eventParser.isStart()) {
eventParser.stop();
afterStopEventParser(eventParser);
}
if (eventSink.isStart()) {
eventSink.stop();
}
if (eventStore.isStart()) {
eventStore.stop();
}
if (metaManager.isStart()) {
metaManager.stop();
}
if (alarmHandler.isStart()) {
alarmHandler.stop();
}
// if (zkClientx != null) {
// zkClientx.close();
// }
super.stop();
logger.info("stop successful....");
}
public boolean subscribeChange(ClientIdentity identity) {
if (StringUtils.isNotEmpty(identity.getFilter())) {
AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
}
} else {
((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
}
}
// filter的处理规则
// a. parser处理数据过滤处理
// b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份,每份的数据可以根据自己的过滤规则不同而有不同的数据
// 后续内存版的一对多分发,可以考虑
return true;
}
protected void afterStartEventParser(CanalEventParser eventParser) {
super.afterStartEventParser(eventParser);
// 读取一下历史订阅的filter信息
List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
for (ClientIdentity clientIdentity : clientIdentitys) {
subscribeChange(clientIdentity);
}
super.start();
}
protected void initAlarmHandler() {
@@ -515,32 +406,6 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
return ZkClientx.getZkClient(StringUtils.join(zkClusters, ";"));
}
// =====================================
public String getDestination() {
return destination;
}
public CanalMetaManager getMetaManager() {
return metaManager;
}
public CanalEventStore<Event> getEventStore() {
return eventStore;
}
public CanalEventParser getEventParser() {
return eventParser;
}
public CanalEventSink<List<Entry>> getEventSink() {
return eventSink;
}
public CanalAlarmHandler getAlarmHandler() {
return alarmHandler;
}
public void setAlarmHandler(CanalAlarmHandler alarmHandler) {
this.alarmHandler = alarmHandler;
}

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -2,20 +2,14 @@ package com.alibaba.otter.canal.instance.spring;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceSupport;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event;
@@ -27,130 +21,13 @@ import com.alibaba.otter.canal.store.model.Event;
* @author zebin.xuzb
* @version 1.0.0
*/
public class CanalInstanceWithSpring extends CanalInstanceSupport implements CanalInstance {
public class CanalInstanceWithSpring extends AbstractCanalInstance {
private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);
private String destination;
private CanalEventParser eventParser;
private CanalEventSink<List<CanalEntry.Entry>> eventSink;
private CanalEventStore<Event> eventStore;
private CanalMetaManager metaManager;
private CanalAlarmHandler alarmHandler;
public String getDestination() {
return this.destination;
}
public CanalEventParser getEventParser() {
return this.eventParser;
}
public CanalEventSink<List<CanalEntry.Entry>> getEventSink() {
return this.eventSink;
}
public CanalEventStore<Event> getEventStore() {
return this.eventStore;
}
public CanalMetaManager getMetaManager() {
return this.metaManager;
}
public CanalAlarmHandler getAlarmHandler() {
return alarmHandler;
}
public boolean subscribeChange(ClientIdentity identity) {
if (StringUtils.isNotEmpty(identity.getFilter())) {
logger.info("subscribe filter change to " + identity.getFilter());
AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
boolean isGroup = (eventParser instanceof GroupEventParser);
if (isGroup) {
// 处理group的模式
List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
}
} else {
((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
}
}
// filter的处理规则
// a. parser处理数据过滤处理
// b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份,每份的数据可以根据自己的过滤规则不同而有不同的数据
// 后续内存版的一对多分发,可以考虑
return true;
}
protected void afterStartEventParser(CanalEventParser eventParser) {
super.afterStartEventParser(eventParser);
// 读取一下历史订阅的filter信息
List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
for (ClientIdentity clientIdentity : clientIdentitys) {
subscribeChange(clientIdentity);
}
}
public void start() {
super.start();
logger.info("start CannalInstance for {}-{} ", new Object[] { 1, destination });
if (!metaManager.isStart()) {
metaManager.start();
}
if (!eventStore.isStart()) {
eventStore.start();
}
if (!eventSink.isStart()) {
eventSink.start();
}
if (!eventParser.isStart()) {
beforeStartEventParser(eventParser);
eventParser.start();
afterStartEventParser(eventParser);
}
logger.info("start successful....");
}
public void stop() {
logger.info("stop CannalInstance for {}-{} ", new Object[] { 1, destination });
if (eventParser.isStart()) {
beforeStopEventParser(eventParser);
eventParser.stop();
afterStopEventParser(eventParser);
}
if (eventSink.isStart()) {
eventSink.stop();
}
if (eventStore.isStart()) {
eventStore.stop();
}
if (metaManager.isStart()) {
metaManager.stop();
}
if (alarmHandler.isStart()) {
alarmHandler.stop();
}
// if (zkClientx != null) {
// zkClientx.close();
// }
super.stop();
logger.info("stop successful....");
super.start();
}
// ======== setter ========

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -142,8 +142,8 @@ public class PeriodMixedMetaManager extends MemoryMetaManager implements CanalMe
}
public void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException {
updateCursorTasks.add(clientIdentity);// 添加到任务队列中进行触发
super.updateCursor(clientIdentity, position);
updateCursorTasks.add(clientIdentity);// 添加到任务队列中进行触发
}
public Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException {

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>canal.parse</artifactId>

View File

@@ -22,7 +22,6 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException;
import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry;
@@ -120,7 +119,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
throw new CanalParseException("consume failed!");
}
LogPosition position = buildLastTranasctionPosition(transaction);
LogPosition position = buildLastTransactionPosition(transaction);
if (position != null) { // 可能position为空
logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
}
@@ -159,7 +158,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
erosaConnection.connect();// 链接
// 4. 获取最后的位置信息
final EntryPosition startPosition = findStartPosition(erosaConnection);
EntryPosition position = findStartPosition(erosaConnection);
final EntryPosition startPosition = position;
if (startPosition == null) {
throw new CanalParseException("can't find start position for " + destination);
}
@@ -190,9 +190,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return running;
} catch (TableIdNotFoundException e) {
throw e;
} catch (Exception e) {
} catch (Throwable e) {
// 记录一下,出错的位点信息
processError(e,
processSinkError(e,
this.lastPosition,
startPosition.getJournalName(),
startPosition.getPosition());
@@ -219,6 +219,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
logger.error(String.format("dump address %s has an error, retrying. caused by ",
runningInfo.getAddress().toString()), e);
} catch (Throwable e) {
processDumpError(e);
exception = e;
if (!running) {
if (!(e instanceof java.nio.channels.ClosedByInterruptException || e.getCause() instanceof java.nio.channels.ClosedByInterruptException)) {
@@ -231,6 +232,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
sendAlarm(destination, ExceptionUtils.getFullStackTrace(e));
}
} finally {
// 重新置为中断状态
Thread.interrupted();
// 关闭一下链接
afterDump(erosaConnection);
try {
@@ -321,7 +324,6 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
startTs = System.currentTimeMillis();
}
CanalEntry.Entry event = binlogParser.parse(bod);
if (enabled) {
this.parsingInterval = System.currentTimeMillis() - startTs;
}
@@ -336,20 +338,11 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return profilingEnabled.get();
}
protected LogPosition buildLastTranasctionPosition(List<CanalEntry.Entry> entries) { // 初始化一下
protected LogPosition buildLastTransactionPosition(List<CanalEntry.Entry> entries) { // 初始化一下
for (int i = entries.size() - 1; i > 0; i--) {
CanalEntry.Entry entry = entries.get(i);
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {// 尽量记录一个事务做为position
LogPosition logPosition = new LogPosition();
EntryPosition position = new EntryPosition();
position.setJournalName(entry.getHeader().getLogfileName());
position.setPosition(entry.getHeader().getLogfileOffset());
position.setTimestamp(entry.getHeader().getExecuteTime());
logPosition.setPostion(position);
LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
logPosition.setIdentity(identity);
return logPosition;
return buildLastPosition(entry);
}
}
@@ -362,6 +355,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
position.setJournalName(entry.getHeader().getLogfileName());
position.setPosition(entry.getHeader().getLogfileOffset());
position.setTimestamp(entry.getHeader().getExecuteTime());
// add serverId at 2016-06-28
position.setServerId(entry.getHeader().getServerId());
logPosition.setPostion(position);
LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
@@ -369,7 +364,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return logPosition;
}
protected void processError(Exception e, LogPosition lastPosition, String startBinlogFile, long startPosition) {
protected void processSinkError(Throwable e, LogPosition lastPosition, String startBinlogFile, long startPosition) {
if (lastPosition != null) {
logger.warn(String.format("ERROR ## parse this event has an error , last position : [%s]",
lastPosition.getPostion()),
@@ -381,13 +376,21 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
}
}
protected void processDumpError(Throwable e) {
// do nothing
}
protected void startHeartBeat(ErosaConnection connection) {
lastEntryTime = 0L; // 初始化
if (timer == null) {// lazy初始化一下
String name = String.format("destination = %s , address = %s , HeartBeatTimeTask",
destination,
runningInfo == null ? null : runningInfo.getAddress().toString());
synchronized (MysqlEventParser.class) {
synchronized (AbstractEventParser.class) {
// synchronized (MysqlEventParser.class) {
// why use MysqlEventParser.class, u know, MysqlEventParser is
// the child class 4 AbstractEventParser,
// do this is ...
if (timer == null) {
timer = new Timer(name, true);
}

View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.parse.inbound;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
@@ -44,6 +45,17 @@ public class TableMeta {
this.fileds = fileds;
}
public List<FieldMeta> getPrimaryFields() {
List<FieldMeta> primarys = new ArrayList<TableMeta.FieldMeta>();
for (FieldMeta meta : fileds) {
if (meta.isKey()) {
primarys.add(meta);
}
}
return primarys;
}
public static class FieldMeta {
private String columnName;

View File

@@ -22,6 +22,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
protected boolean filterQueryDcl = false;
protected boolean filterQueryDml = false;
protected boolean filterQueryDdl = false;
protected boolean filterRows = false;
protected boolean filterTableError = false;
protected BinlogParser buildParser() {
@@ -38,6 +39,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
convert.setFilterQueryDcl(filterQueryDcl);
convert.setFilterQueryDml(filterQueryDml);
convert.setFilterQueryDdl(filterQueryDdl);
convert.setFilterRows(filterRows);
convert.setFilterTableError(filterTableError);
return convert;
}
@@ -87,6 +89,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
this.filterQueryDdl = filterQueryDdl;
}
public void setFilterRows(boolean filterRows) {
this.filterRows = filterRows;
}
public void setFilterTableError(boolean filterTableError) {
this.filterTableError = filterTableError;
}

View File

@@ -8,7 +8,6 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
import com.alibaba.otter.canal.parse.inbound.SinkFunction;
import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue;
@@ -18,6 +17,7 @@ import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import com.taobao.tddl.dbsync.binlog.LogPosition;
import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent;
/**
* local bin log connection (not real connection)
@@ -83,9 +83,9 @@ public class LocalBinLogConnection implements ErosaConnection {
context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
while (running) {
boolean needContinue = true;
LogEvent event;
while (fetcher.fetch()) {
event = decoder.decode(fetcher, context);
LogEvent event = null;
L:while (fetcher.fetch()) {
/*event = decoder.decode(fetcher, context);
if (event == null) {
throw new CanalParseException("parse failed");
}
@@ -93,15 +93,20 @@ public class LocalBinLogConnection implements ErosaConnection {
if (!func.sink(event)) {
needContinue = false;
break;
}
}*/
// do {
// event = decoder.decode(fetcher, context);
// if (event != null && !func.sink(event)) {
// needContinue = false;
// break;
// }
// } while (event != null);
do {
if (event == null) {
event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
} else {
event = decoder.decode(fetcher, context);
}
if (event != null && !func.sink(event)) {
needContinue = false;
break L;
}
} while (event != null);
}
if (needContinue) {// 读取下一个
@@ -145,6 +150,7 @@ public class LocalBinLogConnection implements ErosaConnection {
FileLogFetcher fetcher = new FileLogFetcher(bufferSize);
LogDecoder decoder = new LogDecoder();
decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
decoder.handle(LogEvent.QUERY_EVENT);
decoder.handle(LogEvent.XID_EVENT);
LogContext context = new LogContext();

View File

@@ -1,10 +1,14 @@
package com.alibaba.otter.canal.parse.inbound.mysql;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -21,7 +25,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
// 数据库信息
private AuthenticationInfo masterInfo;
private EntryPosition masterPosition; // binlog信息
private MysqlConnection metaConnection; // 查询meta信息的链接
private TableMetaCache tableMetaCache; // 对应meta
private String directory;
private boolean needWait = false;
private int bufferSize = 16 * 1024;
@@ -33,17 +39,60 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
@Override
protected ErosaConnection buildErosaConnection() {
return buildLocalBinLogConnection();
}
}
@Override
protected void preDump(ErosaConnection connection) {
metaConnection = buildMysqlConnection();
try {
metaConnection.connect();
} catch (IOException e) {
throw new CanalParseException(e);
}
tableMetaCache = new TableMetaCache(metaConnection);
((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
}
public void start() throws CanalParseException {
@Override
protected void afterDump(ErosaConnection connection) {
if (metaConnection != null) {
try {
metaConnection.disconnect();
} catch (IOException e) {
logger.error("ERROR # disconnect meta connection for address:{}", metaConnection.getConnector()
.getAddress(), e);
}
}
}
public void start() throws CanalParseException {
if (runningInfo == null) { // 第一次链接主库
runningInfo = masterInfo;
}
super.start();
}
@Override
public void stop() {
if (metaConnection != null) {
try {
metaConnection.disconnect();
} catch (IOException e) {
logger.error("ERROR # disconnect meta connection for address:{}", metaConnection.getConnector()
.getAddress(), e);
}
}
private ErosaConnection buildLocalBinLogConnection() {
if (tableMetaCache != null) {
tableMetaCache.clearTableMeta();
}
super.stop();
}
private ErosaConnection buildLocalBinLogConnection() {
LocalBinLogConnection connection = new LocalBinLogConnection();
connection.setBufferSize(this.bufferSize);
@@ -53,6 +102,19 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
return connection;
}
private MysqlConnection buildMysqlConnection() {
MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
runningInfo.getUsername(),
runningInfo.getPassword(),
connectionCharsetNumber,
runningInfo.getDefaultDatabaseName());
connection.getConnector().setReceiveBufferSize(64 * 1024);
connection.getConnector().setSendBufferSize(64 * 1024);
connection.getConnector().setSoTimeout(30 * 1000);
connection.setCharset(connectionCharset);
return connection;
}
@Override
protected EntryPosition findStartPosition(ErosaConnection connection) {
// 处理逻辑

View File

@@ -6,6 +6,7 @@ import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -24,6 +25,8 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import com.taobao.tddl.dbsync.binlog.LogPosition;
import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
public class MysqlConnection implements ErosaConnection {
@@ -34,6 +37,7 @@ public class MysqlConnection implements ErosaConnection {
private Charset charset = Charset.forName("UTF-8");
private BinlogFormat binlogFormat;
private BinlogImage binlogImage;
private int binlogChecksum;
public MysqlConnection(){
}
@@ -89,6 +93,7 @@ public class MysqlConnection implements ErosaConnection {
decoder.handle(LogEvent.QUERY_EVENT);
decoder.handle(LogEvent.XID_EVENT);
LogContext context = new LogContext();
context.setLogPosition(new LogPosition(binlogfilename));
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
@@ -105,11 +110,14 @@ public class MysqlConnection implements ErosaConnection {
public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
updateSettings();
loadBinlogChecksum();
sendBinlogDump(binlogfilename, binlogPosition);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
fetcher.start(connector.getChannel());
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
context.setLogPosition(new LogPosition(binlogfilename));
context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
@@ -141,6 +149,8 @@ public class MysqlConnection implements ErosaConnection {
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.write(connector.getChannel(), new ByteBuffer[] { ByteBuffer.wrap(binlogDumpHeader.toBytes()),
ByteBuffer.wrap(cmdBody) });
connector.setDumping(true);
}
public MysqlConnection fork() {
@@ -194,9 +204,12 @@ public class MysqlConnection implements ErosaConnection {
// 如果不设置会出现错误: Slave can not handle replication events with the
// checksum that master is configured to log
// 但也不能乱设置,需要和mysql server的checksum配置一致,不然RotateLogEvent会出现乱码
update("set @master_binlog_checksum= '@@global.binlog_checksum'");
// '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出
update("set @master_binlog_checksum= @@global.binlog_checksum");
} catch (Exception e) {
logger.warn(ExceptionUtils.getFullStackTrace(e));
if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
logger.warn(ExceptionUtils.getFullStackTrace(e));
}
}
try {
@@ -254,6 +267,27 @@ public class MysqlConnection implements ErosaConnection {
}
}
/**
* 获取主库checksum信息
* https://dev.mysql.com/doc/refman/5.6/en/replication-options
* -binary-log.html#option_mysqld_binlog-checksum
*/
private void loadBinlogChecksum() {
ResultSetPacket rs = null;
try {
rs = query("select @master_binlog_checksum");
} catch (IOException e) {
throw new CanalParseException(e);
}
List<String> columnValues = rs.getFieldValues();
if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
} else {
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
}
}
public static enum BinlogFormat {
STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED");

View File

@@ -13,6 +13,7 @@ import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.CanalHASwitchable;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
@@ -28,7 +29,6 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -67,7 +67,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验
private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验
// 心跳检查
// update by yishun.chen,特殊异常处理参数
private int dumpErrorCount = 0; // binlogDump失败异常计数
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
protected ErosaConnection buildErosaConnection() {
return buildMysqlConnection(this.runningInfo);
@@ -92,6 +94,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
for (BinlogFormat supportFormat : supportBinlogFormats) {
if (supportFormat != null && format == supportFormat) {
found = true;
break;
}
}
@@ -106,6 +109,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
for (BinlogImage supportImage : supportBinlogImages) {
if (supportImage != null && image == supportImage) {
found = true;
break;
}
}
@@ -176,10 +180,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
protected void stopHeartBeat() {
TimerTask heartBeatTimerTask = this.heartBeatTimerTask;
super.stopHeartBeat();
if (heartBeatTimerTask != null) {
MysqlConnection mysqlConnection = ((MysqlDetectingTimeTask) heartBeatTimerTask).getMysqlConnection();
try {
mysqlConnection.disconnect();
@@ -324,6 +327,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
return startPosition;
}
protected EntryPosition findEndPosition(ErosaConnection connection) throws IOException {
MysqlConnection mysqlConnection = (MysqlConnection) connection;
EntryPosition endPosition = findEndPosition(mysqlConnection);
return endPosition;
}
protected EntryPosition findStartPositionInternal(ErosaConnection connection) {
MysqlConnection mysqlConnection = (MysqlConnection) connection;
LogPosition logPosition = logPositionManager.getLatestIndexBy(destination);
@@ -384,7 +393,27 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
} else {
if (logPosition.getIdentity().getSourceAddress().equals(mysqlConnection.getConnector().getAddress())) {
logger.warn("prepare to find start position just last position");
if (dumpErrorCountThreshold >= 0 && dumpErrorCount > dumpErrorCountThreshold) {
// binlog定位位点失败,可能有两个原因:
// 1. binlog位点被删除
// 2.vip模式的mysql,发生了主备切换,判断一下serverId是否变化,针对这种模式可以发起一次基于时间戳查找合适的binlog位点
boolean case2 = (standbyInfo == null || standbyInfo.getAddress() == null)
&& logPosition.getPostion().getServerId() != null
&& !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
if (case2) {
long timestamp = logPosition.getPostion().getTimestamp();
long newStartTimestamp = timestamp - fallbackIntervalInSeconds * 1000;
logger.warn("prepare to find start position by last position {}:{}:{}", new Object[] { "", "",
logPosition.getPostion().getTimestamp() });
EntryPosition findPosition = findByStartTimeStamp(mysqlConnection, newStartTimestamp);
// 重新置为一下
dumpErrorCount = 0;
return findPosition;
}
}
// 其余情况
logger.warn("prepare to find start position just last position\n {}",
JsonUtils.marshalToString(logPosition));
return logPosition.getPostion();
} else {
// 针对切换的情况,考虑回退时间
@@ -426,7 +455,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
} catch (Exception e) {
// 上一次记录的poistion可能为一条update/insert/delete变更事件,直接进行dump的话,会缺少tableMap事件,导致tableId未进行解析
processError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
processSinkError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
reDump.set(true);
return false;
}
@@ -460,7 +489,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
lastPosition = buildLastPosition(entry);
} catch (Exception e) {
processError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
processSinkError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
return false;
}
@@ -540,6 +569,22 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
return null;
}
/**
* 查询当前db的serverId信息
*/
private Long findServerId(MysqlConnection mysqlConnection) {
try {
ResultSetPacket packet = mysqlConnection.query("show variables like 'server_id'");
List<String> fields = packet.getFieldValues();
if (CollectionUtils.isEmpty(fields)) {
throw new CanalParseException("command : show variables like 'server_id' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
}
return Long.valueOf(fields.get(1));
} catch (IOException e) {
throw new CanalParseException("command : show variables like 'server_id' has an error!", e);
}
}
/**
* 查询当前的binlog位置
*/
@@ -681,8 +726,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
lastPosition = buildLastPosition(entry);
} catch (Exception e) {
processError(e, lastPosition, searchBinlogFile, 4L);
} catch (Throwable e) {
processSinkError(e, lastPosition, searchBinlogFile, 4L);
}
return running;
@@ -700,21 +745,17 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
}
protected Entry parseAndProfilingIfNecessary(LogEvent bod) throws Exception {
long startTs = -1;
boolean enabled = getProfilingEnabled();
if (enabled) {
startTs = System.currentTimeMillis();
}
CanalEntry.Entry event = binlogParser.parse(bod);
if (enabled) {
this.parsingInterval = System.currentTimeMillis() - startTs;
@Override
protected void processDumpError(Throwable e) {
if (e instanceof IOException) {
String message = e.getMessage();
if (StringUtils.contains(message, "errno = 1236")) {
// 1236 errorCode代表ER_MASTER_FATAL_ERROR_READING_BINLOG
dumpErrorCount++;
}
}
if (parsedEventCount.incrementAndGet() < 0) {
parsedEventCount.set(0);
}
return event;
super.processDumpError(e);
}
public void setSupportBinlogFormats(String formatStrs) {
@@ -800,4 +841,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
public void setHaController(CanalHAController haController) {
this.haController = haController;
}
public void setDumpErrorCountThreshold(int dumpErrorCountThreshold) {
this.dumpErrorCountThreshold = dumpErrorCountThreshold;
}
}

View File

@@ -84,6 +84,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
private boolean filterQueryDdl = false;
// 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92
private boolean filterTableError = false;
// 新增rows过滤,用于仅订阅除rows以外的数据
private boolean filterRows = false;
public Entry parse(LogEvent logEvent) throws CanalParseException {
if (logEvent == null || logEvent instanceof UnknownLogEvent) {
@@ -215,12 +217,20 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
// 更新下table meta cache
if (tableMetaCache != null
&& (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) {
if (StringUtils.isNotEmpty(tableName)) {
// 如果解析到了正确的表信息,则根据全名进行清除
tableMetaCache.clearTableMeta(schemaName, tableName);
} else {
// 如果无法解析正确的表信息,则根据schema进行清除
tableMetaCache.clearTableMetaWithSchemaName(schemaName);
for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) {
String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响
if (StringUtils.isNotEmpty(renameResult.getSchemaName())) {
schemaName0 = renameResult.getSchemaName();
}
tableName = renameResult.getTableName();
if (StringUtils.isNotEmpty(tableName)) {
// 如果解析到了正确的表信息,则根据全名进行清除
tableMetaCache.clearTableMeta(schemaName0, tableName);
} else {
// 如果无法解析正确的表信息,则根据schema进行清除
tableMetaCache.clearTableMetaWithSchemaName(schemaName0);
}
}
}
@@ -297,6 +307,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
private Entry parseRowsEvent(RowsLogEvent event) {
if (filterRows) {
return null;
}
try {
TableMapLogEvent table = event.getTable();
if (table == null) {
@@ -313,6 +326,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return null;
}
if (tableMetaCache.isOnRDS() && "mysql.ha_health_check".equals(fullname)) {
// 忽略rds模式的mysql.ha_health_check心跳数据
return null;
}
EventType eventType = null;
int type = event.getHeader().getType();
if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
@@ -337,7 +355,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
rowChangeBuider.setEventType(eventType);
RowsLogBuffer buffer = event.getRowsBuf(charset.name());
BitSet columns = event.getColumns();
BitSet changeColumns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
boolean tableError = false;
TableMeta tableMeta = null;
if (tableMetaCache != null) {// 入错存在table meta cache
@@ -367,7 +385,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
break;
}
tableError |= parseOneRow(rowDataBuilder, event, buffer, event.getChangeColumns(), true, tableMeta);
tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta);
}
rowChangeBuider.addRowDatas(rowDataBuilder.build());
@@ -389,34 +407,51 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsLogEvent event, RowsLogBuffer buffer, BitSet cols,
boolean isAfter, TableMeta tableMeta) throws UnsupportedEncodingException {
final int columnCnt = event.getTable().getColumnCnt();
final ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
int columnCnt = event.getTable().getColumnCnt();
ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
boolean tableError = false;
// check table fileds count,只能处理加字段
boolean existRDSNoPrimaryKey = false;
if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) {
// online ddl增加字段操作步骤:
// 1. 新增一张临时表,将需要做ddl表的数据全量导入
// 2. 在老表上建立I/U/D的trigger,增量的将数据插入到临时表
// 3. 锁住应用请求,将临时表rename为老表的名字,完成增加字段的操作
// 尝试做一次reload,可能因为ddl没有正确解析,或者使用了类似online ddl的操作
// 因为online ddl没有对应表名的alter语法,所以不会有clear cache的操作
tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false);// 强制重新获取一次
if (tableMeta == null) {
tableError = true;
if (!filterTableError) {
throw new CanalParseException("not found [" + event.getTable().getDbName() + "."
+ event.getTable().getTableName() + "] in db , pls check!");
if (tableMetaCache.isOnRDS()) {
// 特殊处理下RDS的场景
List<FieldMeta> primaryKeys = tableMeta.getPrimaryFields();
if (primaryKeys == null || primaryKeys.isEmpty()) {
if (columnInfo.length == tableMeta.getFileds().size() + 1
&& columnInfo[columnInfo.length - 1].type == LogEvent.MYSQL_TYPE_LONGLONG) {
existRDSNoPrimaryKey = true;
}
}
}
// 在做一次判断
if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) {
tableError = true;
if (!filterTableError) {
throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName() + ","
+ columnInfo.length + " vs " + tableMeta.getFileds().size());
if (!existRDSNoPrimaryKey) {
// online ddl增加字段操作步骤:
// 1. 新增一张临时表,将需要做ddl表的数据全量导入
// 2. 在老表上建立I/U/D的trigger,增量的将数据插入到临时表
// 3. 锁住应用请求,将临时表rename为老表的名字,完成增加字段的操作
// 尝试做一次reload,可能因为ddl没有正确解析,或者使用了类似online ddl的操作
// 因为online ddl没有对应表名的alter语法,所以不会有clear cache的操作
tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false);// 强制重新获取一次
if (tableMeta == null) {
tableError = true;
if (!filterTableError) {
throw new CanalParseException("not found [" + event.getTable().getDbName() + "."
+ event.getTable().getTableName() + "] in db , pls check!");
}
}
// 在做一次判断
if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) {
tableError = true;
if (!filterTableError) {
throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName()
+ "," + columnInfo.length + " vs " + tableMeta.getFileds().size());
}
}
} else {
logger.warn("[" + event.getTable().getDbName() + "." + event.getTable().getTableName()
+ "] is no primary key , skip alibaba_rds_row_id column");
}
}
@@ -427,6 +462,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
continue;
}
if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
// 不解析最后一列
buffer.nextValue(info.type, info.meta, false);
continue;
}
Column.Builder columnBuilder = Column.newBuilder();
FieldMeta fieldMeta = null;
@@ -452,6 +493,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
}
buffer.nextValue(info.type, info.meta, isBinary);
if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
// 不解析最后一列
continue;
}
int javaType = buffer.getJavaType();
if (buffer.isNull()) {
@@ -615,8 +660,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
if (column.getIsNull() && newValue == null) {
// 如果全是null
return false;
} else if (newValue != null && column.getValue().equals(newValue)) {
// 如果不围null,并且相等
} else if (newValue != null && (!column.getIsNull() && column.getValue().equals(newValue))) {
// fixed issue #135, old column is Null
// 如果不为null,并且相等
return false;
}
}
@@ -705,4 +751,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
this.filterTableError = filterTableError;
}
public void setFilterRows(boolean filterRows) {
this.filterRows = filterRows;
}
}

View File

@@ -23,15 +23,17 @@ import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
*/
public class SimpleDdlParser {
public static final String CREATE_PATTERN = "^\\s*CREATE\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
public static final String DROP_PATTERN = "^\\s*DROP\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
public static final String ALERT_PATTERN = "^\\s*ALTER\\s*(IGNORE)?\\s*TABLE\\s*(.*)$";
public static final String TRUNCATE_PATTERN = "^\\s*TRUNCATE\\s*(TABLE)?\\s*(.*)$";
public static final String TABLE_PATTERN = "^(IF\\s*NOT\\s*EXISTS\\s*)?(IF\\s*EXISTS\\s*)?(`?.+?`?[;\\(\\s]+?)?.*$"; // 采用非贪婪模式
public static final String INSERT_PATTERN = "^\\s*(INSERT|MERGE|REPLACE)(.*)$";
public static final String UPDATE_PATTERN = "^\\s*UPDATE(.*)$";
public static final String DELETE_PATTERN = "^\\s*DELETE(.*)$";
public static final String RENAME_PATTERN = "^\\s*RENAME\\s*TABLE\\s*(.*?)\\s*TO\\s*(.*?)$";
public static final String CREATE_PATTERN = "^\\s*CREATE\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
public static final String DROP_PATTERN = "^\\s*DROP\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
public static final String ALERT_PATTERN = "^\\s*ALTER\\s*(IGNORE)?\\s*TABLE\\s*(.*)$";
public static final String TRUNCATE_PATTERN = "^\\s*TRUNCATE\\s*(TABLE)?\\s*(.*)$";
public static final String TABLE_PATTERN = "^(IF\\s*NOT\\s*EXISTS\\s*)?(IF\\s*EXISTS\\s*)?(`?.+?`?[;\\(\\s]+?)?.*$"; // 采用非贪婪模式
public static final String INSERT_PATTERN = "^\\s*(INSERT|MERGE|REPLACE)(.*)$";
public static final String UPDATE_PATTERN = "^\\s*UPDATE(.*)$";
public static final String DELETE_PATTERN = "^\\s*DELETE(.*)$";
public static final String RENAME_PATTERN = "^\\s*RENAME\\s+TABLE\\s+(.+?)\\s+TO\\s+(.+?)$";
public static final String RENAME_REMNANT_PATTERN = "^\\s*(.+?)\\s+TO\\s+(.+?)$";
/**
* <pre>
* CREATE [UNIQUE|FULLTEXT|SPATIAL] INDEX index_name
@@ -42,8 +44,8 @@ public class SimpleDdlParser {
* http://dev.mysql.com/doc/refman/5.6/en/create-index.html
* </pre>
*/
public static final String CREATE_INDEX_PATTERN = "^\\s*CREATE\\s*(UNIQUE)?(FULLTEXT)?(SPATIAL)?\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
public static final String DROP_INDEX_PATTERN = "^\\s*DROP\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
public static final String CREATE_INDEX_PATTERN = "^\\s*CREATE\\s*(UNIQUE)?(FULLTEXT)?(SPATIAL)?\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
public static final String DROP_INDEX_PATTERN = "^\\s*DROP\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
public static DdlResult parse(String queryString, String schmeaName) {
queryString = removeComment(queryString); // 去除/* */的sql注释内容
@@ -74,6 +76,18 @@ public class SimpleDdlParser {
result = parseRename(queryString, schmeaName, RENAME_PATTERN);
if (result != null) {
result.setType(EventType.RENAME);
String[] renameStrings = queryString.split(",");
if (renameStrings.length > 1) {
DdlResult lastResult = result;
for (int i = 1; i < renameStrings.length; i++) {
DdlResult ddlResult = parseRename(renameStrings[i], schmeaName, RENAME_REMNANT_PATTERN);
ddlResult.setType(EventType.RENAME);
lastResult.setRenameTableResult(ddlResult);
lastResult = ddlResult;
}
}
return result;
}
@@ -209,9 +223,15 @@ public class SimpleDdlParser {
private String schemaName;
private String tableName;
private String oriSchemaName; // rename ddl中的源表
private String oriTableName; // rename ddl中的目标表
private String oriSchemaName; // rename ddl中的源表
private String oriTableName; // rename ddl中的目标表
private EventType type;
private DdlResult renameTableResult; // 多个rename table的存储
/*
* RENAME TABLE tbl_name TO new_tbl_name [, tbl_name2 TO new_tbl_name2]
* ...
*/
public DdlResult(){
}
@@ -272,11 +292,28 @@ public class SimpleDdlParser {
this.oriTableName = oriTableName;
}
@Override
public String toString() {
return "DdlResult [schemaName=" + schemaName + ", tableName=" + tableName + ", oriSchemaName="
+ oriSchemaName + ", oriTableName=" + oriTableName + ", type=" + type + "]";
public DdlResult getRenameTableResult() {
return renameTableResult;
}
public void setRenameTableResult(DdlResult renameTableResult) {
this.renameTableResult = renameTableResult;
}
@Override
public String toString() {
DdlResult ddlResult = this;
StringBuffer sb = new StringBuffer();
do {
sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
ddlResult.schemaName,
ddlResult.tableName,
ddlResult.oriSchemaName,
ddlResult.oriTableName,
ddlResult.type));
ddlResult = ddlResult.renameTableResult;
} while (ddlResult != null);
return sb.toString();
}
}
}

View File

@@ -32,6 +32,7 @@ public class TableMetaCache {
public static final String COLUMN_DEFAULT = "COLUMN_DEFAULT";
public static final String EXTRA = "EXTRA";
private MysqlConnection connection;
private boolean isOnRDS = false;
// 第一层tableId,第二层schema.table,解决tableId重复,对应多张表
private Map<String, TableMeta> tableMetaCache;
@@ -56,6 +57,13 @@ public class TableMetaCache {
});
try {
ResultSetPacket packet = connection.query("show global variables like 'rds\\_%'");
if (packet.getFieldValues().size() > 0) {
isOnRDS = true;
}
} catch (IOException e) {
}
}
public TableMeta getTableMeta(String schema, String table) {
@@ -136,4 +144,13 @@ public class TableMetaCache {
.append('`')
.toString();
}
public boolean isOnRDS() {
return isOnRDS;
}
public void setOnRDS(boolean isOnRDS) {
this.isOnRDS = isOnRDS;
}
}

View File

@@ -9,6 +9,8 @@ import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.IOFileFilter;
@@ -188,7 +190,9 @@ public class BinLogFileQueue {
files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
public boolean accept(File file) {
return file.getName().startsWith(baseName);
Pattern pattern = Pattern.compile("\\d+$");
Matcher matcher = pattern.matcher(file.getName());
return file.getName().startsWith(baseName) && matcher.find();
}
public boolean accept(File dir, String name) {

View File

@@ -13,6 +13,7 @@ import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogIdentity;
import com.alibaba.otter.canal.protocol.position.LogPosition;
@@ -22,8 +23,8 @@ public class MysqlEventParserTest {
private static final String DETECTING_SQL = "insert into retl.xdual values(1,now()) on duplicate key update x=now()";
private static final String MYSQL_ADDRESS = "127.0.0.1";
private static final String USERNAME = "xxxxx";
private static final String PASSWORD = "xxxxx";
private static final String USERNAME = "root";
private static final String PASSWORD = "xxxxxx";
@Test
public void test_position() throws InterruptedException {
@@ -45,20 +46,23 @@ public class MysqlEventParserTest {
public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
throws CanalSinkException {
for (Entry entry : entrys) {
entryCount.incrementAndGet();
String logfilename = entry.getHeader().getLogfileName();
long logfileoffset = entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
entryPosition.setJournalName(logfilename);
entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
if (entry.getEntryType() != EntryType.HEARTBEAT) {
entryCount.incrementAndGet();
String logfilename = entry.getHeader().getLogfileName();
long logfileoffset = entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
entryPosition.setJournalName(logfilename);
entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
}
}
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
if (entryCount.get() > 0) {
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
}
return true;
}
});
@@ -92,14 +96,14 @@ public class MysqlEventParserTest {
@Test
public void test_timestamp() throws InterruptedException {
final TimeoutChecker timeoutChecker = new TimeoutChecker(30 * 1000);
final TimeoutChecker timeoutChecker = new TimeoutChecker(3000 * 1000);
final AtomicLong entryCount = new AtomicLong(0);
final EntryPosition entryPosition = new EntryPosition();
final MysqlEventParser controller = new MysqlEventParser();
final EntryPosition defaultPosition = buildPosition(null, null, 1322803601000L);
final EntryPosition defaultPosition = buildPosition(null, null, 1475116855000L);
controller.setSlaveId(3344L);
controller.setDetectingEnable(true);
controller.setDetectingEnable(false);
controller.setDetectingSQL(DETECTING_SQL);
controller.setMasterInfo(buildAuthentication());
controller.setMasterPosition(defaultPosition);
@@ -109,22 +113,25 @@ public class MysqlEventParserTest {
public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
throws CanalSinkException {
for (Entry entry : entrys) {
entryCount.incrementAndGet();
if (entry.getEntryType() != EntryType.HEARTBEAT) {
entryCount.incrementAndGet();
String logfilename = entry.getHeader().getLogfileName();
long logfileoffset = entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
String logfilename = entry.getHeader().getLogfileName();
long logfileoffset = entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
entryPosition.setJournalName(logfilename);
entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
entryPosition.setJournalName(logfilename);
entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
}
}
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
if (entryCount.get() > 0) {
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
}
return true;
}
});
@@ -174,21 +181,26 @@ public class MysqlEventParserTest {
public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
throws CanalSinkException {
for (Entry entry : entrys) {
entryCount.incrementAndGet();
if (entry.getEntryType() != EntryType.HEARTBEAT) {
String logfilename = entry.getHeader().getLogfileName();
long logfileoffset = entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
entryCount.incrementAndGet();
entryPosition.setJournalName(logfilename);
entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
String logfilename = entry.getHeader().getLogfileName();
long logfileoffset = entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
entryPosition.setJournalName(logfilename);
entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
}
}
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
if (entryCount.get() > 0) {
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
}
return true;
}
});
@@ -244,22 +256,27 @@ public class MysqlEventParserTest {
public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
throws CanalSinkException {
for (Entry entry : entrys) {
entryCount.incrementAndGet();
if (entry.getEntryType() != EntryType.HEARTBEAT) {
entryCount.incrementAndGet();
// String logfilename = entry.getHeader().getLogfileName();
// long logfileoffset =
// entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
// String logfilename =
// entry.getHeader().getLogfileName();
// long logfileoffset =
// entry.getHeader().getLogfileOffset();
long executeTime = entry.getHeader().getExecuteTime();
// entryPosition.setJournalName(logfilename);
// entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
// entryPosition.setJournalName(logfilename);
// entryPosition.setPosition(logfileoffset);
entryPosition.setTimestamp(executeTime);
break;
}
}
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
if (entryCount.get() > 0) {
controller.stop();
timeoutChecker.stop();
timeoutChecker.touch();
}
return true;
}
});

View File

@@ -149,6 +149,61 @@ public class SimpleDdlParserTest {
Assert.assertEquals("retl2", result.getSchemaName());
Assert.assertEquals("retl_mark", result.getOriTableName());
Assert.assertEquals("retl_mark2", result.getTableName());
queryString = "rename \n table \n `retl`.`retl_mark` to `retl2.retl_mark2` , `retl1`.`retl_mark1` to `retl3.retl_mark3`;";
result = SimpleDdlParser.parse(queryString, "retl");
Assert.assertNotNull(result);
Assert.assertEquals("retl", result.getOriSchemaName());
Assert.assertEquals("retl2", result.getSchemaName());
Assert.assertEquals("retl_mark", result.getOriTableName());
Assert.assertEquals("retl_mark2", result.getTableName());
result = result.getRenameTableResult();
Assert.assertNotNull(result);
Assert.assertEquals("retl1", result.getOriSchemaName());
Assert.assertEquals("retl3", result.getSchemaName());
Assert.assertEquals("retl_mark1", result.getOriTableName());
Assert.assertEquals("retl_mark3", result.getTableName());
//正则匹配test case
queryString = "rename table totl_mark to totl_mark2";
result = SimpleDdlParser.parse(queryString, "retl");
Assert.assertNotNull(result);
Assert.assertEquals("retl", result.getOriSchemaName());
Assert.assertEquals("retl", result.getSchemaName());
Assert.assertEquals("totl_mark", result.getOriTableName());
Assert.assertEquals("totl_mark2", result.getTableName());
queryString = "rename table totl.retl_mark to totl2.retl_mark2";
result = SimpleDdlParser.parse(queryString, "retl");
Assert.assertNotNull(result);
Assert.assertEquals("totl", result.getOriSchemaName());
Assert.assertEquals("totl2", result.getSchemaName());
Assert.assertEquals("retl_mark", result.getOriTableName());
Assert.assertEquals("retl_mark2", result.getTableName());
queryString = "rename \n table \n `totl`.`retl_mark` to `totl2.retl_mark2`;";
result = SimpleDdlParser.parse(queryString, "retl");
Assert.assertNotNull(result);
Assert.assertEquals("totl", result.getOriSchemaName());
Assert.assertEquals("totl2", result.getSchemaName());
Assert.assertEquals("retl_mark", result.getOriTableName());
Assert.assertEquals("retl_mark2", result.getTableName());
queryString = "rename \n table \n `totl`.`retl_mark` to `totl2.retl_mark2` , `totl1`.`retl_mark1` to `totl3.retl_mark3`;";
result = SimpleDdlParser.parse(queryString, "retl");
Assert.assertNotNull(result);
Assert.assertEquals("totl", result.getOriSchemaName());
Assert.assertEquals("totl2", result.getSchemaName());
Assert.assertEquals("retl_mark", result.getOriTableName());
Assert.assertEquals("retl_mark2", result.getTableName());
result = result.getRenameTableResult();
Assert.assertNotNull(result);
Assert.assertEquals("totl1", result.getOriSchemaName());
Assert.assertEquals("totl3", result.getSchemaName());
Assert.assertEquals("retl_mark1", result.getOriTableName());
Assert.assertEquals("retl_mark3", result.getTableName());
}
@Test

View File

@@ -4,7 +4,7 @@
<artifactId>canal</artifactId>
<packaging>pom</packaging>
<name>canal module for otter ${project.version}</name>
<version>1.0.20</version>
<version>1.0.23</version>
<url>https://github.com/alibaba/canal</url>
<parent>
<groupId>org.sonatype.oss</groupId>
@@ -202,7 +202,7 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>2.4.1</version>
<version>2.6.1</version>
</dependency>
<!-- log -->
<dependency>

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -15,6 +15,8 @@ public class EntryPosition extends TimePosition {
private boolean included = false;
private String journalName;
private Long position;
// add by agapple at 2016-06-28
private Long serverId = null; // 记录一下位点对应的serverId
public EntryPosition(){
super(null);
@@ -34,6 +36,11 @@ public class EntryPosition extends TimePosition {
this.position = position;
}
public EntryPosition(String journalName, Long position, Long timestamp, Long serverId){
this(journalName, position, timestamp);
this.serverId = serverId;
}
public String getJournalName() {
return journalName;
}
@@ -58,6 +65,14 @@ public class EntryPosition extends TimePosition {
this.included = included;
}
public Long getServerId() {
return serverId;
}
public void setServerId(Long serverId) {
this.serverId = serverId;
}
@Override
public int hashCode() {
final int prime = 31;

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>canal.server</artifactId>

View File

@@ -21,6 +21,7 @@ import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.protocol.position.PositionRange;
import com.alibaba.otter.canal.server.CanalServer;
import com.alibaba.otter.canal.server.CanalService;
import com.alibaba.otter.canal.server.exception.CanalServerException;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.model.Event;
@@ -37,24 +38,39 @@ import com.google.common.collect.MigrateMap;
* @author zebin.xuzb
* @version 1.0.0
*/
public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, com.alibaba.otter.canal.server.CanalService {
public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService {
private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
private Map<String, CanalInstance> canalInstances;
// private Map<ClientIdentity, Position> lastRollbackPostions;
private CanalInstanceGenerator canalInstanceGenerator;
private static class SingletonHolder {
private static final CanalServerWithEmbedded CANAL_SERVER_WITH_EMBEDDED = new CanalServerWithEmbedded();
}
public CanalServerWithEmbedded(){
// 希望也保留用户new单独实例的需求,兼容历史
}
public static CanalServerWithEmbedded instance() {
return SingletonHolder.CANAL_SERVER_WITH_EMBEDDED;
}
public void start() {
super.start();
if (!isStart()) {
super.start();
canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>() {
canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>() {
public CanalInstance apply(String destination) {
return canalInstanceGenerator.generate(destination);
}
});
public CanalInstance apply(String destination) {
return canalInstanceGenerator.generate(destination);
}
});
// lastRollbackPostions = new MapMaker().makeMap();
// lastRollbackPostions = new MapMaker().makeMap();
}
}
public void stop() {
@@ -115,6 +131,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
*/
@Override
public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
checkStart(clientIdentity.getDestination());
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
if (!canalInstance.getMetaManager().isStart()) {
canalInstance.getMetaManager().start();

View File

@@ -33,11 +33,17 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
private Channel serverChannel = null;
private ServerBootstrap bootstrap = null;
public CanalServerWithNetty(){
private static class SingletonHolder {
private static final CanalServerWithNetty CANAL_SERVER_WITH_NETTY = new CanalServerWithNetty();
}
public CanalServerWithNetty(CanalServerWithEmbedded embeddedServer){
this.embeddedServer = embeddedServer;
private CanalServerWithNetty(){
this.embeddedServer = CanalServerWithEmbedded.instance();
}
public static CanalServerWithNetty instance() {
return SingletonHolder.CANAL_SERVER_WITH_NETTY;
}
public void start() {

View File

@@ -66,7 +66,6 @@ public class SessionHandler extends SimpleChannelHandler {
Short.valueOf(sub.getClientId()),
sub.getFilter());
MDC.put("destination", clientIdentity.getDestination());
embeddedServer.subscribe(clientIdentity);
// 尝试启动,如果已经启动,忽略
if (!embeddedServer.isStart(clientIdentity.getDestination())) {
@@ -76,6 +75,7 @@ public class SessionHandler extends SimpleChannelHandler {
}
}
embeddedServer.subscribe(clientIdentity);
ctx.setAttachment(clientIdentity);// 设置状态数据
NettyUtils.ack(ctx.getChannel(), null);
} else {

View File

@@ -31,7 +31,7 @@ public abstract class BaseCanalServerWithEmbededTest {
@Before
public void setUp() {
server = new CanalServerWithEmbedded();
server = CanalServerWithEmbedded.instance();
server.setCanalInstanceGenerator(new CanalInstanceGenerator() {
public CanalInstance generate(String destination) {

View File

@@ -58,7 +58,8 @@ public class CanalServerWithNettyTest {
}
});
nettyServer = new CanalServerWithNetty(embeddedServer);
nettyServer = CanalServerWithNetty.instance();
nettyServer.setEmbeddedServer(embeddedServer);
nettyServer.setPort(1088);
nettyServer.start();
}

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.20</version>
<version>1.0.23</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>

View File

@@ -51,6 +51,8 @@ public class CanalEventUtils {
position.setJournalName(event.getEntry().getHeader().getLogfileName());
position.setPosition(event.getEntry().getHeader().getLogfileOffset());
position.setTimestamp(event.getEntry().getHeader().getExecuteTime());
// add serverId at 2016-06-28
position.setServerId(event.getEntry().getHeader().getServerId());
LogPosition logPosition = new LogPosition();
logPosition.setPostion(position);