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

fixed semi sync format & refactor

Browse Source
This commit is contained in:
agapple
2017-12-27 16:53:09 +08:00
parent c6f4de9f88
commit 9595e6abbd
7 changed files with 81 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java
View File

@@ -20,7 +20,7 @@ public class LogBuffer {
protected int origin, limit; protected int origin, limit;
protected int position; protected int position;
protected int semival; protected int semival;
protected LogBuffer(){ protected LogBuffer(){
} }

10
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java
View File

@@ -104,12 +104,14 @@ public final class LogDecoder {
try { try {
/* Decoding binary-log to event */ /* Decoding binary-log to event */
event = decode(buffer, header, context); event = decode(buffer, header, context);
if(event!=null){ if (event != null) {
event.setSemival(buffer.semival); event.setSemival(buffer.semival);
} }
} catch (IOException e) { } catch (IOException e) {
if (logger.isWarnEnabled()) logger.warn("Decoding " + LogEvent.getTypeName(header.getType()) if (logger.isWarnEnabled()) {
+ " failed from: " + context.getLogPosition(), e); logger.warn("Decoding " + LogEvent.getTypeName(header.getType()) + " failed from: "
+ context.getLogPosition(), e);
}
throw e; throw e;
} finally { } finally {
buffer.limit(limit); /* Restore limit */ buffer.limit(limit); /* Restore limit */

9
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java
View File

@@ -361,9 +361,14 @@ public abstract class LogEvent {
protected final LogHeader header; protected final LogHeader header;
/** /**
* mysql半同步semi标识 0不需要semi ack 给mysql, 1需要semi ack给mysql * mysql半同步semi标识
*
* <pre>
* 0不需要semi ack 给mysql
* 1需要semi ack给mysql
* </pre>
*/ */
protected int semival; protected int semival;

41
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/MariaGtidLogEvent.java
View File

@@ -13,35 +13,30 @@ import com.taobao.tddl.dbsync.binlog.event.LogHeader;
*/ */
public class MariaGtidLogEvent extends IgnorableLogEvent { public class MariaGtidLogEvent extends IgnorableLogEvent {
private long gtid; 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 /**
* <pre>
uint<8> commit_id * mariadb gtidlog event format
* uint<8> GTID sequence
else * uint<4> Replication Domain ID
* uint<1> Flags
uint<6> 0 *
* if flag & FL_GROUP_COMMIT_ID
* uint<8> commit_id
* else
* uint<6> 0
* </pre>
*/
*
*/
public MariaGtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ public MariaGtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
super(header, buffer, descriptionEvent); super(header, buffer, descriptionEvent);
gtid=buffer.getUlong64().longValue(); gtid = buffer.getUlong64().longValue();
// do nothing , just ignore log event // do nothing , just ignore log event
} }
public long getGtid() { public long getGtid() {
return gtid; return gtid;
} }
} }

53
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/SemiAckCommandPacket.java
View File

@@ -10,26 +10,23 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
/** /**
* semi ack command * semi ack command
*
* @author amos_chen * @author amos_chen
*
*/ */
public class SemiAckCommandPacket extends CommandPacket { public class SemiAckCommandPacket extends CommandPacket {
public long binlogPosition; public long binlogPosition;
public String binlogFileName;
public String binlogFileName;
public SemiAckCommandPacket(){
}
@Override
public void fromBytes(byte[] data) throws IOException {
// TODO Auto-generated method stub
} public SemiAckCommandPacket(){
/** }
@Override
public void fromBytes(byte[] data) throws IOException {
}
/**
* <pre> * <pre>
* Bytes Name * Bytes Name
* -------------------------------------------------------- * --------------------------------------------------------
@@ -37,22 +34,22 @@ public class SemiAckCommandPacket extends CommandPacket {
* ----- ---- * ----- ----
* 1 semi mark * 1 semi mark
* 8 binlog position to start at (little endian) * 8 binlog position to start at (little endian)
* n binlog file name * n binlog file name
* *
* </pre> * </pre>
*/ */
public byte[] toBytes() throws IOException { public byte[] toBytes() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream(); ByteArrayOutputStream out = new ByteArrayOutputStream();
//0 write semi mark // 0 write semi mark
out.write(0xef); out.write(0xef);
//1 write 8 bytes for position // 1 write 8 bytes for position
ByteHelper.write8ByteUnsignedIntLittleEndian(binlogPosition, out); ByteHelper.write8ByteUnsignedIntLittleEndian(binlogPosition, out);
//2 write binlog filename // 2 write binlog filename
if(StringUtils.isNotEmpty(binlogFileName)){ if (StringUtils.isNotEmpty(binlogFileName)) {
out.write(binlogFileName.getBytes()); out.write(binlogFileName.getBytes());
} }
return out.toByteArray(); return out.toByteArray();
} }
} }

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

@@ -25,7 +25,6 @@ import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent; import com.taobao.tddl.dbsync.binlog.LogEvent;
import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent;
public class MysqlConnection implements ErosaConnection { public class MysqlConnection implements ErosaConnection {
@@ -37,8 +36,6 @@ public class MysqlConnection implements ErosaConnection {
private BinlogFormat binlogFormat; private BinlogFormat binlogFormat;
private BinlogImage binlogImage; private BinlogImage binlogImage;
private String binlogfilename;
// tsdb releated // tsdb releated
private AuthenticationInfo authInfo; private AuthenticationInfo authInfo;
protected int connTimeout = 5 * 1000; // 5秒 protected int connTimeout = 5 * 1000; // 5秒
@@ -145,18 +142,13 @@ public class MysqlConnection implements ErosaConnection {
if (event == null) { if (event == null) {
throw new CanalParseException("parse failed"); throw new CanalParseException("parse failed");
} }
//binlog日志文件发生变化
if(event.getHeader().getType()==LogEvent.ROTATE_EVENT){
binlogfilename=((RotateLogEvent)event).getFilename();
}
if (!func.sink(event)) { if (!func.sink(event)) {
break; break;
}else{ }
if(event.getSemival()==1){
sendSemiAck(binlogfilename, binlogPosition); if (event.getSemival() == 1) {
} sendSemiAck(context.getLogPosition().getFileName(), binlogPosition);
} }
} }
} }
@@ -180,12 +172,12 @@ public class MysqlConnection implements ErosaConnection {
PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody); PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
connector.setDumping(true); connector.setDumping(true);
} }
private void sendSemiAck(String binlogfilename, Long binlogPosition) throws IOException { private void sendSemiAck(String binlogfilename, Long binlogPosition) throws IOException {
SemiAckCommandPacket semiAckCmd = new SemiAckCommandPacket(); SemiAckCommandPacket semiAckCmd = new SemiAckCommandPacket();
semiAckCmd.binlogFileName = binlogfilename; semiAckCmd.binlogFileName = binlogfilename;
semiAckCmd.binlogPosition = binlogPosition; semiAckCmd.binlogPosition = binlogPosition;
byte[] cmdBody = semiAckCmd.toBytes(); byte[] cmdBody = semiAckCmd.toBytes();
logger.info("SEMI ACK with position:{}", semiAckCmd); logger.info("SEMI ACK with position:{}", semiAckCmd);
@@ -193,7 +185,7 @@ public class MysqlConnection implements ErosaConnection {
semiAckHeader.setPacketBodyLength(cmdBody.length); semiAckHeader.setPacketBodyLength(cmdBody.length);
semiAckHeader.setPacketSequenceNumber((byte) 0x00); semiAckHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.writePkg(connector.getChannel(), semiAckHeader.toBytes(), cmdBody); PacketManager.writePkg(connector.getChannel(), semiAckHeader.toBytes(), cmdBody);
} }
public MysqlConnection fork() { public MysqlConnection fork() {

34
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
View File

@@ -36,8 +36,8 @@ public class DirectLogFetcher extends LogFetcher {
public static final int MAX_PACKET_LENGTH = (256 * 256 * 256 - 1); public static final int MAX_PACKET_LENGTH = (256 * 256 * 256 - 1);
private SocketChannel channel; private SocketChannel channel;
private boolean issemi=false; private boolean issemi = false;
// private BufferedInputStream input; // private BufferedInputStream input;
@@ -55,9 +55,9 @@ public class DirectLogFetcher extends LogFetcher {
public void start(SocketChannel channel) throws IOException { public void start(SocketChannel channel) throws IOException {
this.channel = channel; this.channel = channel;
String dbsemi= System.getProperty("db.semi"); String dbsemi = System.getProperty("db.semi");
if("1".equals(dbsemi)){ if ("1".equals(dbsemi)) {
issemi=true; issemi = true;
} }
// 和mysql driver一样，提供buffer机制，提升读取binlog速度 // 和mysql driver一样，提供buffer机制，提升读取binlog速度
// this.input = new // this.input = new
@@ -111,15 +111,15 @@ public class DirectLogFetcher extends LogFetcher {
+ ", len = " + netlen); + ", len = " + netlen);
} }
} }
//if mysql is in semi mode // if mysql is in semi mode
if(issemi){ if (issemi) {
//parse semi mark // parse semi mark
int semimark=getUint8(NET_HEADER_SIZE+1); int semimark = getUint8(NET_HEADER_SIZE + 1);
int semival=getUint8(NET_HEADER_SIZE+2); int semival = getUint8(NET_HEADER_SIZE + 2);
this.semival=semival; this.semival = semival;
} }
// The first packet is a multi-packet, concatenate the packets. // The first packet is a multi-packet, concatenate the packets.
while (netlen == MAX_PACKET_LENGTH) { while (netlen == MAX_PACKET_LENGTH) {
if (!fetch0(0, NET_HEADER_SIZE)) { if (!fetch0(0, NET_HEADER_SIZE)) {
@@ -136,10 +136,10 @@ public class DirectLogFetcher extends LogFetcher {
} }
// Preparing buffer variables to decoding. // Preparing buffer variables to decoding.
if(issemi){ if (issemi) {
origin = NET_HEADER_SIZE + 3; origin = NET_HEADER_SIZE + 3;
}else{ } else {
origin = NET_HEADER_SIZE + 1; origin = NET_HEADER_SIZE + 1;
} }
position = origin; position = origin;
limit -= origin; limit -= origin;