From d66839c965d9070ee00061619fff97e55e0bbaf4 Mon Sep 17 00:00:00 2001 From: lucifax301 <1918831771@qq.com> Date: Mon, 18 Dec 2017 14:10:53 +0800 Subject: [PATCH] mysql semi support and mariadb gtid parse --- .../taobao/tddl/dbsync/binlog/LogBuffer.java | 1 + .../taobao/tddl/dbsync/binlog/LogDecoder.java | 3 + .../taobao/tddl/dbsync/binlog/LogEvent.java | 17 +++++- .../event/mariadb/MariaGtidLogEvent.java | 26 +++++++++ .../packets/client/SemiAckCommandPacket.java | 58 +++++++++++++++++++ .../parse/driver/mysql/utils/ByteHelper.java | 13 ++++- .../parse/inbound/mysql/MysqlConnection.java | 28 +++++++++ .../mysql/dbsync/DirectLogFetcher.java | 22 ++++++- 8 files changed, 164 insertions(+), 4 deletions(-) create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/SemiAckCommandPacket.java diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java index f81660fa..5557bea7 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java @@ -20,6 +20,7 @@ public class LogBuffer { protected int origin, limit; protected int position; + protected int semival; protected LogBuffer(){ } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java index 95a57424..0f9c32b5 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java @@ -104,6 +104,9 @@ public final class LogDecoder { try { /* Decoding binary-log to event */ event = decode(buffer, header, context); + if(event!=null){ + event.setSemival(buffer.semival); + } } catch (IOException e) { if (logger.isWarnEnabled()) logger.warn("Decoding " + LogEvent.getTypeName(header.getType()) + " failed from: " + context.getLogPosition(), e); diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java index 4dba80ee..d8abf4c1 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java @@ -359,8 +359,23 @@ public abstract class LogEvent { protected static final Log logger = LogFactory.getLog(LogEvent.class); protected final LogHeader header; + + /** + * mysql半同步semi标识 0不需要semi ack 给mysql, 1需要semi ack给mysql + */ + protected int semival; + + - protected LogEvent(LogHeader header){ + public int getSemival() { + return semival; + } + + public void setSemival(int semival) { + this.semival = semival; + } + + protected LogEvent(LogHeader header){ this.header = header; } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/MariaGtidLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/MariaGtidLogEvent.java index 9dc22b39..6b00a633 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/MariaGtidLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/MariaGtidLogEvent.java @@ -13,9 +13,35 @@ import com.taobao.tddl.dbsync.binlog.event.LogHeader; */ public class MariaGtidLogEvent extends IgnorableLogEvent { + private long gtid; + + /** + * mariadb gtidlog event format + * + uint<8> GTID sequence + uint<4> Replication Domain ID + uint<1> Flags + + if flag & FL_GROUP_COMMIT_ID + + uint<8> commit_id + + else + + uint<6> 0 + + * + */ + public MariaGtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ super(header, buffer, descriptionEvent); + gtid=buffer.getUlong64().longValue(); // do nothing , just ignore log event } + public long getGtid() { + return gtid; + } + + } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/SemiAckCommandPacket.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/SemiAckCommandPacket.java new file mode 100644 index 00000000..f96238b3 --- /dev/null +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/SemiAckCommandPacket.java @@ -0,0 +1,58 @@ +package com.alibaba.otter.canal.parse.driver.mysql.packets.client; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +import org.apache.commons.lang.StringUtils; + +import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket; +import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper; + +/** + * semi ack command + * @author amos_chen + * + */ +public class SemiAckCommandPacket extends CommandPacket { + + public long binlogPosition; + + public String binlogFileName; + + public SemiAckCommandPacket(){ + + } + + @Override + public void fromBytes(byte[] data) throws IOException { + // TODO Auto-generated method stub + + } + + /** + * 
+     * Bytes                        Name
+     *  --------------------------------------------------------
+     *  Bytes                        Name
+     *  -----                        ----
+     *  1                            semi mark
+     *  8                            binlog position to start at (little endian)
+     *  n                            binlog file name 
+     * 
+     *
+ */ + public byte[] toBytes() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + //0 write semi mark + out.write(0xef); + //1 write 8 bytes for position + ByteHelper.write8ByteUnsignedIntLittleEndian(binlogPosition, out); + + //2 write binlog filename + if(StringUtils.isNotEmpty(binlogFileName)){ + out.write(binlogFileName.getBytes()); + } + return out.toByteArray(); + } + +} diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/ByteHelper.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/ByteHelper.java index 2c7cd31b..67c3830e 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/ByteHelper.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/ByteHelper.java @@ -108,7 +108,18 @@ public abstract class ByteHelper { return out.toByteArray(); } - + + public static void write8ByteUnsignedIntLittleEndian(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 writeUnsignedIntLittleEndian(long data, ByteArrayOutputStream out) { out.write((byte) (data & 0xFF)); out.write((byte) (data >>> 8)); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 7b141010..ff9c3718 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -14,6 +14,7 @@ import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor; import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.client.SemiAckCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; import com.alibaba.otter.canal.parse.exception.CanalParseException; @@ -24,6 +25,7 @@ import com.alibaba.otter.canal.parse.support.AuthenticationInfo; 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.event.RotateLogEvent; public class MysqlConnection implements ErosaConnection { @@ -35,6 +37,8 @@ public class MysqlConnection implements ErosaConnection { private BinlogFormat binlogFormat; private BinlogImage binlogImage; + private String binlogfilename; + // tsdb releated private AuthenticationInfo authInfo; protected int connTimeout = 5 * 1000; // 5秒 @@ -141,9 +145,18 @@ public class MysqlConnection implements ErosaConnection { if (event == null) { throw new CanalParseException("parse failed"); } + + //binlog日志文件发生变化 + if(event.getHeader().getType()==LogEvent.ROTATE_EVENT){ + binlogfilename=((RotateLogEvent)event).getFilename(); + } if (!func.sink(event)) { break; + }else{ + if(event.getSemival()==1){ + sendSemiAck(binlogfilename, binlogPosition); + } } } } @@ -167,6 +180,21 @@ public class MysqlConnection implements ErosaConnection { PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody); connector.setDumping(true); } + + private void sendSemiAck(String binlogfilename, Long binlogPosition) throws IOException { + SemiAckCommandPacket semiAckCmd = new SemiAckCommandPacket(); + semiAckCmd.binlogFileName = binlogfilename; + semiAckCmd.binlogPosition = binlogPosition; + + byte[] cmdBody = semiAckCmd.toBytes(); + + logger.info("SEMI ACK with position:{}", semiAckCmd); + HeaderPacket semiAckHeader = new HeaderPacket(); + semiAckHeader.setPacketBodyLength(cmdBody.length); + semiAckHeader.setPacketSequenceNumber((byte) 0x00); + PacketManager.writePkg(connector.getChannel(), semiAckHeader.toBytes(), cmdBody); + + } public MysqlConnection fork() { MysqlConnection connection = new MysqlConnection(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java index ad0e95e7..8827a87a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java @@ -36,6 +36,8 @@ public class DirectLogFetcher extends LogFetcher { public static final int MAX_PACKET_LENGTH = (256 * 256 * 256 - 1); private SocketChannel channel; + + private boolean issemi=false; // private BufferedInputStream input; @@ -53,6 +55,10 @@ public class DirectLogFetcher extends LogFetcher { public void start(SocketChannel channel) throws IOException { this.channel = channel; + String dbsemi= System.getProperty("db.semi"); + if("1".equals(dbsemi)){ + issemi=true; + } // 和mysql driver一样，提供buffer机制，提升读取binlog速度 // this.input = new // BufferedInputStream(channel.socket().getInputStream(), 16384); @@ -105,7 +111,15 @@ public class DirectLogFetcher extends LogFetcher { + ", len = " + netlen); } } - + + //if mysql is in semi mode + if(issemi){ + //parse semi mark + int semimark=getUint8(NET_HEADER_SIZE+1); + int semival=getUint8(NET_HEADER_SIZE+2); + this.semival=semival; + } + // The first packet is a multi-packet, concatenate the packets. while (netlen == MAX_PACKET_LENGTH) { if (!fetch0(0, NET_HEADER_SIZE)) { @@ -122,7 +136,11 @@ public class DirectLogFetcher extends LogFetcher { } // Preparing buffer variables to decoding. - origin = NET_HEADER_SIZE + 1; + if(issemi){ + origin = NET_HEADER_SIZE + 3; + }else{ + origin = NET_HEADER_SIZE + 1; + } position = origin; limit -= origin; return true;