chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #618 from hiwjd/feature/gtid-support

Browse Source 
支持通过GTID同步binlog
This commit is contained in:
agapple
2018-05-07 10:30:58 +08:00
committed by GitHub
parent 5c2ed38d2f 21f3f8c973
commit 912a7a12be
22 changed files with 13479 additions and 12615 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java
View File

@@ -3,6 +3,9 @@ package com.taobao.tddl.dbsync.binlog.event;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import java.nio.ByteBuffer;
import java.util.UUID;
/**
* @author jianghang 2013-4-8 上午12:36:29
* @version 1.0.3
@@ -16,6 +19,8 @@ public class GtidLogEvent extends LogEvent {
public static final int ENCODED_SID_LENGTH = 16;
private boolean commitFlag;
private UUID sid;
private long gno;
public GtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
super(header);
@@ -27,6 +32,14 @@ public class GtidLogEvent extends LogEvent {
buffer.position(commonHeaderLen);
commitFlag = (buffer.getUint8() != 0); // ENCODED_FLAG_LENGTH
byte[] bs = buffer.getData(ENCODED_SID_LENGTH);
ByteBuffer bb = ByteBuffer.wrap(bs);
long high = bb.getLong();
long low = bb.getLong();
sid = new UUID(high, low);
gno = buffer.getLong64();
// ignore gtid info read
// sid.copy_from((uchar *)ptr_buffer);
// ptr_buffer+= ENCODED_SID_LENGTH;
@@ -42,4 +55,11 @@ public class GtidLogEvent extends LogEvent {
return commitFlag;
}
public UUID getSid() {
return sid;
}
public long getGno() {
return gno;
}
}

5
deployer/src/main/resources/example/instance.properties
View File

@@ -6,7 +6,9 @@ canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
canal.instance.gtidon=true
# table meta tsdb info
canal.instance.tsdb.enable=true
@@ -20,7 +22,8 @@ canal.instance.tsdb.dbPassword=canal
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

5
deployer/src/main/resources/spring/file-instance.xml
View File

@@ -151,6 +151,7 @@
<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}" />
<property name="gtid" value="${canal.instance.master.gtid}" />
</bean>
</property>
<property name="standbyPosition">
@@ -158,6 +159,7 @@
<property name="journalName" value="${canal.instance.standby.journal.name}" />
<property name="position" value="${canal.instance.standby.position}" />
<property name="timestamp" value="${canal.instance.standby.timestamp}" />
<property name="gtid" value="${canal.instance.standby.gtid}" />
</bean>
</property>
<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
@@ -172,5 +174,8 @@
<!--表结构相关-->
<property name="enableTsdb" value="${canal.instance.tsdb.enable:true}"/>
<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
<!--是否启用GTID模式-->
<property name="isGTIDMode" value="${canal.instance.gtidon}"/>
</bean>
</beans>

24
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/GTIDSet.java Normal file
View File

@@ -0,0 +1,24 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets;
import java.io.IOException;
/**
* Created by hiwjd on 2018/4/23.
* hiwjd0@gmail.com
*/
public interface GTIDSet {
/**
* 序列化成字节数组
*
* @return
*/
byte[] encode() throws IOException;
/**
* 更新当前实例
* @param str
* @throws Exception
*/
void update(String str);
}

151
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MysqlGTIDSet.java Normal file
View File

@@ -0,0 +1,151 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.*;
/**
* Created by hiwjd on 2018/4/23.
* hiwjd0@gmail.com
*/
public class MysqlGTIDSet implements GTIDSet {
public Map<String, UUIDSet> sets;
@Override
public byte[] encode() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
ByteHelper.writeUnsignedInt64LittleEndian(sets.size(), out);
for (Map.Entry<String, UUIDSet> entry : sets.entrySet()) {
out.write(entry.getValue().encode());
}
return out.toByteArray();
}
@Override
public void update(String str) {
UUIDSet us = UUIDSet.parse(str);
String sid = us.SID.toString();
if (sets.containsKey(sid)) {
sets.get(sid).intervals.addAll(us.intervals);
sets.get(sid).intervals = UUIDSet.combine(sets.get(sid).intervals);
} else {
sets.put(sid, us);
}
}
@Override
public boolean equals(Object o) {
if (o == null) return false;
if (this == o) return true;
MysqlGTIDSet gs = (MysqlGTIDSet) o;
if (gs.sets == null) return false;
for (Map.Entry<String, UUIDSet> entry : sets.entrySet()) {
if (!entry.getValue().equals(gs.sets.get(entry.getKey()))) {
return false;
}
}
return true;
}
/**
* 解析如下格式的字符串为MysqlGTIDSet:
*
* 726757ad-4455-11e8-ae04-0242ac110002:1 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:2}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:4}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:4 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:5}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:4}, {start:7, stop: 10}]
* }
* }
* }
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4 =>
* MysqlGTIDSet{
* sets: {
* 726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:1, stop:4}]
* },
* 726757ad-4455-11e8-ae04-0242ac110003: UUIDSet{
* SID: 726757ad-4455-11e8-ae04-0242ac110002,
* intervals: [{start:4, stop:5}]
* }
* }
* }
*
* @param gtidData
* @return
*/
public static MysqlGTIDSet parse(String gtidData) {
Map<String, UUIDSet> m;
if (gtidData == null || gtidData.length() < 1) {
m = new HashMap<String, UUIDSet>();
} else {
String[] uuidStrs = gtidData.split(",");
m = new HashMap<String, UUIDSet>(uuidStrs.length);
for (int i = 0; i < uuidStrs.length; i++) {
UUIDSet uuidSet = UUIDSet.parse(uuidStrs[i]);
m.put(uuidSet.SID.toString(), uuidSet);
}
}
MysqlGTIDSet gs = new MysqlGTIDSet();
gs.sets = m;
return gs;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
for (Map.Entry<String, UUIDSet> entry : sets.entrySet()) {
if (sb.length() > 0) {
sb.append(",");
}
sb.append(entry.getValue().toString());
}
return sb.toString();
}
}

202
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java Normal file
View File

@@ -0,0 +1,202 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
/**
* Created by hiwjd on 2018/4/23.
* hiwjd0@gmail.com
*/
public class UUIDSet {
public UUID SID;
public List<Interval> intervals;
public byte[] encode() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
ByteBuffer bb = ByteBuffer.wrap(new byte[16]);
bb.putLong(SID.getMostSignificantBits());
bb.putLong(SID.getLeastSignificantBits());
out.write(bb.array());
ByteHelper.writeUnsignedInt64LittleEndian(intervals.size(), out);
for (Interval interval : intervals) {
ByteHelper.writeUnsignedInt64LittleEndian(interval.start, out);
ByteHelper.writeUnsignedInt64LittleEndian(interval.stop, out);
}
return out.toByteArray();
}
@Override
public boolean equals(Object o) {
if (o == null) return false;
if (this == o) return true;
UUIDSet us = (UUIDSet) o;
Collections.sort(intervals);
Collections.sort(us.intervals);
if (SID.equals(us.SID) && intervals.equals(us.intervals)) {
return true;
}
return false;
}
public static class Interval implements Comparable<Interval> {
public long start;
public long stop;
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Interval interval = (Interval) o;
if (start != interval.start) return false;
return stop == interval.stop;
}
@Override
public int hashCode() {
int result = (int) (start ^ (start >>> 32));
result = 31 * result + (int) (stop ^ (stop >>> 32));
return result;
}
@Override
public int compareTo(Interval o) {
if (equals(o)) {
return 1;
}
return start > o.start ? 1 : (start == o.start ? 0 : -1);
}
}
/**
* 解析如下格式字符串为UUIDSet:
*
* 726757ad-4455-11e8-ae04-0242ac110002:1 =>
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:2}]}
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3 =>
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}]}
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:4
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:5}]}
*
* 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9
* UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}, {start:7, stop:10}]}
*
* @param str
* @return
*/
public static UUIDSet parse(String str) {
String[] ss = str.split(":");
if (ss.length < 2) {
throw new RuntimeException(String.format("parseUUIDSet failed due to wrong format: %s", str));
}
List<Interval> intervals = new ArrayList<Interval>();
for (int i=1; i<ss.length; i++) {
intervals.add(parseInterval(ss[i]));
}
UUIDSet uuidSet = new UUIDSet();
uuidSet.SID = UUID.fromString(ss[0]);
uuidSet.intervals = combine(intervals);
return uuidSet;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(SID.toString());
for (Interval interval : intervals) {
if (interval.start == interval.stop - 1) {
sb.append(":");
sb.append(interval.start);
} else {
sb.append(":");
sb.append(interval.start);
sb.append("-");
sb.append(interval.stop-1);
}
}
return sb.toString();
}
/**
* 解析如下格式字符串为Interval:
*
* 1 => Interval{start:1, stop:2}
* 1-3 => Interval{start:1, stop:4}
*
* 注意！字符串格式表达时[n,m]是两侧都包含的，Interval表达时[n,m)右侧开
*
* @param str
* @return
*/
public static Interval parseInterval(String str) {
String[] ss = str.split("-");
Interval interval = new Interval();
switch (ss.length) {
case 1:
interval.start = Long.parseLong(ss[0]);
interval.stop = interval.start + 1;
break;
case 2:
interval.start = Long.parseLong(ss[0]);
interval.stop = Long.parseLong(ss[1]) + 1;
break;
default:
throw new RuntimeException(String.format("parseInterval failed due to wrong format: %s", str));
}
return interval;
}
/**
* 把{start,stop}连续的合并掉:
* [{start:1, stop:4},{start:4, stop:5}] => [{start:1, stop:5}]
*
* @param intervals
* @return
*/
public static List<Interval> combine(List<Interval> intervals) {
List<Interval> combined = new ArrayList<Interval>();
Collections.sort(intervals);
int len = intervals.size();
for (int i=0; i<len; i++) {
combined.add(intervals.get(i));
int j;
for (j=i+1; j<len; j++) {
if (intervals.get(i).stop >= intervals.get(j).start) {
intervals.get(i).stop = intervals.get(j).stop;
} else {
break;
}
}
i = j - 1;
}
return combined;
}
}

68
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/BinlogDumpGTIDCommandPacket.java Normal file
View File

@@ -0,0 +1,68 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
/**
* Created by hiwjd on 2018/4/24.
* hiwjd0@gmail.com
*
* https://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html
*/
public class BinlogDumpGTIDCommandPacket extends CommandPacket {
public static final int BINLOG_DUMP_NON_BLOCK = 0x01;
public static final int BINLOG_THROUGH_POSITION = 0x02;
public static final int BINLOG_THROUGH_GTID = 0x04;
public long slaveServerId;
public GTIDSet gtidSet;
public BinlogDumpGTIDCommandPacket() {
setCommand((byte) 0x1e);
}
@Override
public void fromBytes(byte[] data) throws IOException {
}
@Override
public byte[] toBytes() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
// 0. [1] write command number
out.write(getCommand());
// 1. [2] flags
ByteHelper.writeUnsignedShortLittleEndian(BINLOG_THROUGH_GTID, out);
// 2. [4] server-id
ByteHelper.writeUnsignedIntLittleEndian(slaveServerId, out);
// 3. [4] binlog-filename-len
ByteHelper.writeUnsignedIntLittleEndian(0, out);
// 4. [] binlog-filename
// skip
// 5. [8] binlog-pos
ByteHelper.writeUnsignedInt64LittleEndian(4, out);
// if flags & BINLOG_THROUGH_GTID {
byte[] bs = gtidSet.encode();
// 6. [4] data-size
ByteHelper.writeUnsignedIntLittleEndian(bs.length, out);
// 7, [] data
// [8] n_sids // 文档写的是4个字节，其实是8个字节
// for n_sids {
// [16] SID
// [8] n_intervals
// for n_intervals {
// [8] start (signed)
// [8] end (signed)
// }
// }
out.write(bs);
// }
return out.toByteArray();
}
}

11
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/ByteHelper.java
View File

@@ -127,6 +127,17 @@ public abstract class ByteHelper {
out.write((byte) (data >>> 24));
}
public static void writeUnsignedInt64LittleEndian(long data, ByteArrayOutputStream out) {
out.write((byte) (data & 0xFF));
out.write((byte) (data >>> 8));
out.write((byte) (data >>> 16));
out.write((byte) (data >>> 24));
out.write((byte) (data >>> 32));
out.write((byte) (data >>> 40));
out.write((byte) (data >>> 48));
out.write((byte) (data >>> 56));
}
public static void writeUnsignedShortLittleEndian(int data, ByteArrayOutputStream out) {
out.write((byte) (data & 0xFF));
out.write((byte) ((data >>> 8) & 0xFF));

129
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlGTIDSetTest.java Normal file
View File

@@ -0,0 +1,129 @@
package com.alibaba.otter.canal.parse.driver.mysql;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
import org.junit.Test;
import static org.junit.Assert.*;
import java.io.IOException;
import java.util.*;
/**
* Created by hiwjd on 2018/4/25.
* hiwjd0@gmail.com
*/
public class MysqlGTIDSetTest {
@Test
public void testEncode() throws IOException {
GTIDSet gtidSet = MysqlGTIDSet.parse("726757ad-4455-11e8-ae04-0242ac110002:1");
byte[] bytes = gtidSet.encode();
byte[] expected = {
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x72, 0x67, 0x57, (byte)0xad,
0x44, 0x55, 0x11, (byte)0xe8, (byte)0xae, 0x04, 0x02, 0x42, (byte)0xac, 0x11, 0x00, 0x02,
0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00
};
for (int i=0; i<bytes.length; i++) {
assertEquals(expected[i], bytes[i]);
}
}
@Test
public void testParse() {
Map<String, MysqlGTIDSet> cases = new HashMap<String, MysqlGTIDSet>(5);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 2))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3:4",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 5))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9",
buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4, 7, 10))
);
cases.put(
"726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4",
buildForTest(
Arrays.asList(
new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4),
new Material("726757ad-4455-11e8-ae04-0242ac110003", 4, 5)
)
)
);
for (Map.Entry<String, MysqlGTIDSet> entry : cases.entrySet()) {
MysqlGTIDSet expected = entry.getValue();
MysqlGTIDSet actual = MysqlGTIDSet.parse(entry.getKey());
assertEquals(expected, actual);
}
}
private static class Material {
public Material(String uuid, long start, long stop) {
this.uuid = uuid;
this.start = start;
this.stop = stop;
this.start1 = 0;
this.stop1 = 0;
}
public Material(String uuid, long start, long stop, long start1, long stop1) {
this.uuid = uuid;
this.start = start;
this.stop = stop;
this.start1 = start1;
this.stop1 = stop1;
}
public String uuid;
public long start;
public long stop;
public long start1;
public long stop1;
}
private MysqlGTIDSet buildForTest(Material material) {
return buildForTest(Arrays.asList(material));
}
private MysqlGTIDSet buildForTest(List<Material> materials) {
Map<String, UUIDSet> sets = new HashMap<String, UUIDSet>();
for (Material a : materials) {
UUIDSet.Interval interval = new UUIDSet.Interval();
interval.start = a.start;
interval.stop = a.stop;
List<UUIDSet.Interval> intervals = new ArrayList<UUIDSet.Interval>();
intervals.add(interval);
if (a.start1 > 0 && a.stop1 > 0) {
UUIDSet.Interval interval1 = new UUIDSet.Interval();
interval1.start = a.start1;
interval1.stop = a.stop1;
intervals.add(interval1);
}
UUIDSet us = new UUIDSet();
us.SID = UUID.fromString(a.uuid);
us.intervals = intervals;
sets.put(a.uuid, us);
}
MysqlGTIDSet gs = new MysqlGTIDSet();
gs.sets = sets;
return gs;
}
}

30
driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/UUIDSetTest.java Normal file
View File

@@ -0,0 +1,30 @@
package com.alibaba.otter.canal.parse.driver.mysql;
import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.*;
/**
* Created by hiwjd on 2018/4/26.
* hiwjd0@gmail.com
*/
public class UUIDSetTest {
@Test
public void testToString() {
Map<String, String> cases = new HashMap<String, String>(4);
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1", "726757ad-4455-11e8-ae04-0242ac110002:1");
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3", "726757ad-4455-11e8-ae04-0242ac110002:1-3");
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3:4-6", "726757ad-4455-11e8-ae04-0242ac110002:1-6");
cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3:5-7", "726757ad-4455-11e8-ae04-0242ac110002:1-3:5-7");
for (Map.Entry<String, String> entry : cases.entrySet()) {
String expected = entry.getValue();
assertEquals(expected, UUIDSet.parse(entry.getKey()).toString());
}
}
}

39
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
View File

@@ -8,6 +8,8 @@ import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang.math.RandomUtils;
@@ -87,6 +89,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
protected TimerTask heartBeatTimerTask;
protected Throwable exception = null;
protected boolean isGTIDMode = false; // 是否是GTID模式
protected String GTIDSetStr = null;
protected abstract BinlogParser buildParser();
protected abstract ErosaConnection buildErosaConnection();
@@ -214,12 +219,17 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
};
// 4. 开始dump数据
if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
// 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据
if (isGTIDMode()) {
erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), sinkHandler);
} else {
erosaConnection.dump(startPosition.getJournalName(),
startPosition.getPosition(),
sinkHandler);
if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
} else {
erosaConnection.dump(startPosition.getJournalName(),
startPosition.getPosition(),
sinkHandler);
}
}
} catch (TableIdNotFoundException e) {
@@ -368,6 +378,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
position.setTimestamp(entry.getHeader().getExecuteTime());
// add serverId at 2016-06-28
position.setServerId(entry.getHeader().getServerId());
// set gtid
position.setGtid(entry.getHeader().getGtid());
logPosition.setPostion(position);
LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
@@ -530,4 +543,20 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return exception;
}
public boolean isGTIDMode() {
return isGTIDMode;
}
public void setIsGTIDMode(boolean isGTIDMode) {
this.isGTIDMode = isGTIDMode;
}
public String getGTIDSetStr() {
return GTIDSetStr;
}
public void setGTIDSetStr(String GTIDSetStr) {
this.GTIDSetStr = GTIDSetStr;
}
}

11
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java
View File

@@ -1,5 +1,7 @@
package com.alibaba.otter.canal.parse.inbound;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import java.io.IOException;
/**
@@ -24,5 +26,14 @@ public interface ErosaConnection {
public void dump(long timestamp, SinkFunction func) throws IOException;
/**
* 通过GTID同步binlog
*
* @param gtidSet
* @param func
* @throws IOException
*/
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException;
ErosaConnection fork();
}

18
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
View File

@@ -1,7 +1,10 @@
package com.alibaba.otter.canal.parse.inbound.mysql;
import java.io.IOException;
import java.nio.charset.Charset;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvertGTID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,7 +37,20 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
protected boolean useDruidDdlFilter = true;
protected BinlogParser buildParser() {
LogEventConvert convert = new LogEventConvert();
LogEventConvert convert;
if (isGTIDMode()) {
EntryPosition position;
try {
position = findStartPosition(null);
} catch (IOException e) {
throw new RuntimeException("findStartPosition failed.", e);
}
convert = new LogEventConvertGTID(MysqlGTIDSet.parse(position.getGtid()));
} else {
convert = new LogEventConvert();
}
if (eventFilter != null && eventFilter instanceof AviaterRegexFilter) {
convert.setNameFilter((AviaterRegexFilter) eventFilter);
}

7
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java
View File

@@ -4,6 +4,8 @@ import java.io.File;
import java.io.IOException;
import java.util.List;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -200,6 +202,11 @@ public class LocalBinLogConnection implements ErosaConnection {
dump(binlogFilename, binlogFileOffset, func);
}
@Override
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
throw new NotImplementedException();
}
public ErosaConnection fork() {
LocalBinLogConnection connection = new LocalBinLogConnection();

39
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -8,6 +8,8 @@ import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -159,6 +161,29 @@ public class MysqlConnection implements ErosaConnection {
}
}
@Override
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
updateSettings();
sendBinlogDumpGTID(gtidSet);
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();
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
if (event == null) {
throw new CanalParseException("parse failed");
}
if (!func.sink(event)) {
break;
}
}
}
public void dump(long timestamp, SinkFunction func) throws IOException {
throw new NullPointerException("Not implement yet");
}
@@ -222,6 +247,20 @@ public class MysqlConnection implements ErosaConnection {
PacketManager.writePkg(connector.getChannel(), semiAckHeader.toBytes(), cmdBody);
}
private void sendBinlogDumpGTID(GTIDSet gtidSet) throws IOException {
BinlogDumpGTIDCommandPacket binlogDumpCmd = new BinlogDumpGTIDCommandPacket();
binlogDumpCmd.slaveServerId = this.slaveId;
binlogDumpCmd.gtidSet = gtidSet;
byte[] cmdBody = binlogDumpCmd.toBytes();
logger.info("COM_BINLOG_DUMP_GTID:{}", binlogDumpCmd);
HeaderPacket binlogDumpHeader = new HeaderPacket();
binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
connector.setDumping(true);
}
public MysqlConnection fork() {
MysqlConnection connection = new MysqlConnection();
connection.setCharset(getCharset());

12
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
View File

@@ -10,6 +10,10 @@ import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.inbound.BinlogParser;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvertGTID;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;
@@ -344,6 +348,14 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException {
if (isGTIDMode()) {
// GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的
LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination);
if (logPosition != null) {
return logPosition.getPostion();
}
return masterPosition;
}
EntryPosition startPosition = findStartPositionInternal(connection);
if (needTransactionPosition.get()) {
logger.warn("prepare to find last position : {}", startPosition.toString());

12
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
View File

@@ -56,6 +56,7 @@ import com.taobao.tddl.dbsync.binlog.event.UserVarLogEvent;
import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent;
/**
* 基于{@linkplain LogEvent}转化为Entry对象的处理
@@ -128,6 +129,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
case LogEvent.RAND_EVENT:
return parseRandLogEvent((RandLogEvent) logEvent);
case LogEvent.GTID_LOG_EVENT:
return parseGTIDLogEvent((GtidLogEvent) logEvent);
default:
break;
}
@@ -143,6 +146,15 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
}
private Entry parseGTIDLogEvent(GtidLogEvent logEvent) {
LogHeader logHeader = logEvent.getHeader();
Header header = createHeader("", logHeader, "", "", EventType.GTID);
Pair.Builder builder = Pair.newBuilder();
builder.setKey("gtid");
builder.setValue(String.format("%s:%d", logEvent.getSid(), logEvent.getGno()));
return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
}
private Entry parseQueryEvent(QueryLogEvent event, boolean isSeek) {
String queryString = event.getQuery();
if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {

55
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvertGTID.java Normal file
View File

@@ -0,0 +1,55 @@
package com.alibaba.otter.canal.parse.inbound.mysql.dbsync;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.BinlogParser;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Created by hiwjd on 2018/5/2.
* hiwjd0@gmail.com
*/
public class LogEventConvertGTID extends LogEventConvert implements BinlogParser<LogEvent> {
public static final Logger logger = LoggerFactory.getLogger(LogEventConvertGTID.class);
// latest gtid
private GTIDSet gtidSet;
public LogEventConvertGTID(GTIDSet gtidSet) {
this.gtidSet = gtidSet;
}
@Override
public CanalEntry.Entry parse(LogEvent event, boolean isSeek) throws CanalParseException {
CanalEntry.Entry entry = super.parse(event, isSeek);
if (entry == null) {
return null;
}
if (entry.getEntryType() == CanalEntry.EntryType.GTIDLOG) {
try {
CanalEntry.Pair pair = CanalEntry.Pair.parseFrom(entry.getStoreValue());
gtidSet.update(pair.getValue());
} catch (InvalidProtocolBufferException e) {
logger.error("retrive gtid failed.", e);
throw new CanalParseException("retrive gtid failed.", e);
}
return null;
}
if (gtidSet != null) {
String gtid = gtidSet.toString();
CanalEntry.Header header = entry.getHeader().toBuilder().setGtid(gtid).build();
entry = entry.toBuilder().setHeader(header).build();
}
return entry;
}
}

25235
protocol/src/main/java/com/alibaba/otter/canal/protocol/CanalEntry.java
View File

File diff suppressed because it is too large Load Diff

9
protocol/src/main/java/com/alibaba/otter/canal/protocol/EntryProtocol.proto
View File

@@ -55,7 +55,10 @@ message Header {
optional EventType eventType = 11 [default = UPDATE];
/**预留扩展**/
repeated Pair props = 12;
repeated Pair props = 12;
/**当前事务的gitd**/
optional string gtid = 13;
}
/**每个字段的数据结构**/
@@ -169,7 +172,8 @@ enum EntryType{
ROWDATA = 2;
TRANSACTIONEND = 3;
/** 心跳类型，内部使用，外部暂不可见，可忽略 **/
HEARTBEAT = 4;
HEARTBEAT = 4;
GTIDLOG = 5;
}
/** 事件类型 **/
@@ -186,6 +190,7 @@ enum EventType {
/**CREATE INDEX**/
CINDEX = 10;
DINDEX = 11;
GTID = 12;
}
/**数据库类型**/

9
protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java
View File

@@ -18,6 +18,7 @@ public class EntryPosition extends TimePosition {
private Long position;
// add by agapple at 2016-06-28
private Long serverId = null; // 记录一下位点对应的serverId
private String gtid = null;
public EntryPosition(){
super(null);
@@ -74,6 +75,14 @@ public class EntryPosition extends TimePosition {
this.serverId = serverId;
}
public String getGtid() {
return gtid;
}
public void setGtid(String gtid) {
this.gtid = gtid;
}
@Override
public int hashCode() {
final int prime = 31;

3
store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java
View File

@@ -54,6 +54,9 @@ public class CanalEventUtils {
// add serverId at 2016-06-28
position.setServerId(event.getEntry().getHeader().getServerId());
// add gtid
position.setGtid(event.getEntry().getHeader().getGtid());
LogPosition logPosition = new LogPosition();
logPosition.setPostion(position);
logPosition.setIdentity(event.getLogIdentity());