canal-mysql binlog日志解析
1.参考
【alibaba canal】https://github.com/alibaba/canal
【mysql binlog】https://dev.mysql.com/doc/refman/8.0/en/binary-log.html
【mysql event】https://dev.mysql.com/doc/internals/en/event-meanings.html
【mysqlbinlog 命令】https://dev.mysql.com/doc/refman/8.0/en/mysqlbinlog.html
2.基本简介
canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费
2.1 mysql主备复制实现
- master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；
- slave将master的binary log events拷贝到它的中继日志(relay log)；
- slave重做中继日志中的事件，将改变反映它自己的数据。
2.2 mysql binglog 数据包格式
mysql 事件体目前有一下版本
-
v1：在 MySQL 3.23 中使用
+=====================================+ | event | timestamp 0 : 4 | | header +----------------------------+ | | type_code 4 : 1 | | +----------------------------+ | | server_id 5 : 4 | | +----------------------------+ | | event_length 9 : 4 | +=====================================+ | event | fixed part 13 : y | | data +----------------------------+ | | variable part | +=====================================+ 标头长度 = 13 字节 数据长度 = (event_length - 13) 字节 y 特定于事件类型。 -
v3：在 MySQL 4.0.2 到 4.1 中使用
+=====================================+ | event | timestamp 0 : 4 | | header +----------------------------+ | | type_code 4 : 1 | | +----------------------------+ | | server_id 5 : 4 | | +----------------------------+ | | event_length 9 : 4 | | +----------------------------+ | | next_position 13 : 4 | | +----------------------------+ | | flags 17 : 2 | +=====================================+ | event | fixed part 19 : y | | data +----------------------------+ | | variable part | +=====================================+ 标头长度 = 19 字节 数据长度 = (event_length - 19) 字节 y 特定于事件类型。 -
v4：用于 MySQL 5.0 及更高版本
+=====================================+ | event | timestamp 0 : 4 | | header +----------------------------+ | | type_code 4 : 1 | | +----------------------------+ | | server_id 5 : 4 | | +----------------------------+ | | event_length 9 : 4 | | +----------------------------+ | | next_position 13 : 4 | | +----------------------------+ | | flags 17 : 2 | | +----------------------------+ | | extra_headers 19 : x-19 | +=====================================+ | event | fixed part x : y | | data +----------------------------+ | | variable part | +=====================================+ 标头长度 = x 字节 数据长度 = (event_length - x) 字节 固定数据长度 = y 字节可变数据长度 = (event_length - (x + y)) 字节
2.2 canal工作原理
- canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
- mysql master收到dump请求，开始推送binary log给slave(也就是canal)
- canal解析binary log对象(原始为byte流)
2.3 canal架构
说明：
- server代表一个canal运行实例，对应于一个jvm
- instance对应于一个数据队列 （1个server对应1..n个instance)
instance模块：
-
eventParser (数据源接入，模拟slave协议和master进行交互，协议解析)
-
eventSink (Parser和Store链接器，进行数据过滤，加工，分发的工作)
-
eventStore (数据存储)
-
metaManager (增量订阅&消费信息管理器)
canal-instance
instance代表了一个实际运行的数据队列，包括了EventPaser,EventSink,EventStore等组件。
抽象了CanalInstanceGenerator，主要是考虑配置的管理方式：
- manager方式： 和你自己的内部web console/manager系统进行对接。(目前主要是公司内部使用)
- spring方式：基于spring xml + properties进行定义，构建spring配置.
canal-event-parser
整个parser过程大致可分为几步：
- Connection获取上一次解析成功的位置 (如果第一次启动，则获取初始指定的位置或者是当前数据库的binlog位点)
- Connection建立链接，发送BINLOG_DUMP指令 // 0. write command number // 1. write 4 bytes bin-log position to start at // 2. write 2 bytes bin-log flags // 3. write 4 bytes server id of the slave // 4. write bin-log file name
- Mysql开始推送Binaly Log
- 接收到的Binaly Log的通过Binlog parser进行协议解析，补充一些特定信息 // 补充字段名字，字段类型，主键信息，unsigned类型处理
- 传递给EventSink模块进行数据存储，是一个阻塞操作，直到存储成功
- 存储成功后，定时记录Binaly Log位置
canal-event-sink
说明：
- 数据过滤：支持通配符的过滤模式，表名，字段内容等
- 数据路由/分发：解决1:n (1个parser对应多个store的模式)
- 数据归并：解决n:1 (多个parser对应1个store)
- 数据加工：在进入store之前进行额外的处理，比如join
数据1:n业务
为了合理的利用数据库资源， 一般常见的业务都是按照schema进行隔离，然后在mysql上层或者dao这一层面上，进行一个数据源路由，屏蔽数据库物理位置对开发的影响，阿里系主要是通过cobar/tddl来解决数据源路由问题。
所以，一般一个数据库实例上，会部署多个schema，每个schema会有由1个或者多个业务方关注
数据n:1业务
同样，当一个业务的数据规模达到一定的量级后，必然会涉及到水平拆分和垂直拆分的问题，针对这些拆分的数据需要处理时，就需要链接多个store进行处理，消费的位点就会变成多份，而且数据消费的进度无法得到尽可能有序的保证。
所以，在一定业务场景下，需要将拆分后的增量数据进行归并处理，比如按照时间戳/全局id进行排序归并.
canal-event-store
- 目前仅实现了Memory内存模式，后续计划增加本地file存储，mixed混合模式
- 借鉴了Disruptor的RingBuffer的实现思路
RingBuffer设计：
定义了3个cursor
- Put : Sink模块进行数据存储的最后一次写入位置
- Get : 数据订阅获取的最后一次提取位置
- Ack : 数据消费成功的最后一次消费位置
借鉴Disruptor的RingBuffer的实现，将RingBuffer拉直来看：
实现说明：
- Put/Get/Ack cursor用于递增，采用long型存储
- buffer的get操作，通过取余或者与操作。(与操作： cusor & (size - 1) , size需要为2的指数，效率比较高)
2.4 canal HA设计
canal的ha分为两部分，canal server和canal client分别有对应的ha实现
- canal server: 为了减少对mysql dump的请求，不同server上的instance要求同一时间只能有一个处于running，其他的处于standby状态.
- canal client: 为了保证有序性，一份instance同一时间只能由一个canal client进行get/ack/rollback操作，否则客户端接收无法保证有序。
整个HA机制的控制主要是依赖了zookeeper的几个特性，watcher和EPHEMERAL节点(和session生命周期绑定)。
大致步骤：
- canal server要启动某个canal instance时都先向zookeeper进行一次尝试启动判断 (实现：创建EPHEMERAL节点，谁创建成功就允许谁启动)
- 创建zookeeper节点成功后，对应的canal server就启动对应的canal instance，没有创建成功的canal instance就会处于standby状态
- 一旦zookeeper发现canal server A创建的节点消失后，立即通知其他的canal server再次进行步骤1的操作，重新选出一个canal server启动instance.
- canal client每次进行connect时，会首先向zookeeper询问当前是谁启动了canal instance，然后和其建立链接，一旦链接不可用，会重新尝试connect.
Canal Client的方式和canal server方式类似，也是利用zookeeper的抢占EPHEMERAL节点的方式进行控制.
2.5 canal client
client-server交互
get/ack/rollback协议介绍：
- Message getWithoutAck(int batchSize)，允许指定batchSize，一次可以获取多条，每次返回的对象为Message，包含的内容为： a. batch id 唯一标识 b. entries 具体的数据对象，对应的数据对象格式：EntryProtocol.proto
- void rollback(long batchId)，顾命思议，回滚上次的get请求，重新获取数据。基于get获取的batchId进行提交，避免误操作
- void ack(long batchId)，顾命思议，确认已经消费成功，通知server删除数据。基于get获取的batchId进行提交，避免误操作
canal的get/ack/rollback协议和常规的jms协议有所不同，允许get/ack异步处理，比如可以连续调用get多次，后续异步按顺序提交ack/rollback，项目中称之为流式api.
流式api设计的好处：
- get/ack异步化，减少因ack带来的网络延迟和操作成本 (99%的状态都是处于正常状态，异常的rollback属于个别情况，没必要为个别的case牺牲整个性能)
- get获取数据后，业务消费存在瓶颈或者需要多进程/多线程消费时，可以不停的轮询get数据，不停的往后发送任务，提高并行化. (作者在实际业务中的一个case：业务数据消费需要跨中美网络，所以一次操作基本在200ms以上，为了减少延迟，所以需要实施并行化)
流式api设计：
- 每次get操作都会在meta中产生一个mark，mark标记会递增，保证运行过程中mark的唯一性
- 每次的get操作，都会在上一次的mark操作记录的cursor继续往后取，如果mark不存在，则在last ack cursor继续往后取
- 进行ack时，需要按照mark的顺序进行数序ack，不能跳跃ack. ack会删除当前的mark标记，并将对应的mark位置更新为last ack cusor
- 一旦出现异常情况，客户端可发起rollback情况，重新置位：删除所有的mark, 清理get请求位置，下次请求会从last ack cursor继续往后取
数据对象格式
Entry
Header
logfileName [binlog文件名]
logfileOffset [binlog position]
executeTime [binlog里记录变更发生的时间戳]
schemaName [数据库实例]
tableName [表名]
eventType [insert/update/delete类型]
entryType [事务头BEGIN/事务尾END/数据ROWDATA]
storeValue [byte数据,可展开，对应的类型为RowChange]
RowChange
isDdl [是否是ddl变更操作，比如create table/drop table]
sql [具体的ddl sql]
rowDatas [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]
beforeColumns [Column类型的数组]
afterColumns [Column类型的数组]
Column
index [column序号]
sqlType [jdbc type]
name [column name]
isKey [是否为主键]
updated [是否发生过变更]
isNull [值是否为null]
value [具体的内容，注意为文本]
说明：
- 可以提供数据库变更前和变更后的字段内容，针对binlog中没有的name,isKey等信息进行补全
- 可以提供ddl的变更语句
3.cannel 源码
3.1 canal启动
public void start() {
super.start();
if (!embeddedServer.isStart()) {
embeddedServer.start();
}
//1.创建canal server
this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
bootstrap.setOption("child.keepAlive", true);
bootstrap.setOption("child.tcpNoDelay", true);
//2.构造对应的pipeline
bootstrap.setPipelineFactory(() -> {
ChannelPipeline pipelines = Channels.pipeline();
pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
// support to maintain child socket channel.
pipelines.addLast(HandshakeInitializationHandler.class.getName(),
new HandshakeInitializationHandler(childGroups));
pipelines.addLast(ClientAuthenticationHandler.class.getName(),
new ClientAuthenticationHandler(embeddedServer));
//会话处理
SessionHandler sessionHandler = new SessionHandler(embeddedServer);
pipelines.addLast(SessionHandler.class.getName(), sessionHandler);
return pipelines;
});
//3.启动 默认监听1111端口
if (StringUtils.isNotEmpty(ip)) {
this.serverChannel = bootstrap.bind(new InetSocketAddress(this.ip, this.port));
} else {
this.serverChannel = bootstrap.bind(new InetSocketAddress(this.port));
}
}
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
ClientIdentity clientIdentity = null;
switch (packet.getType()) {
case SUBSCRIPTION:
break;
case UNSUBSCRIPTION:
break;
case GET:
break;
case CLIENTACK:
break;
case CLIENTROLLBACK:
break;
default:
break;
}
}
3.2 client代码
3.2.1 订阅数据-socket
public static void main(String args[]) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), "example", "", "");
int batchSize = 1000;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
while (isRunning) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(batchSize);
long batchId = message.getId();
int size = message.getEntries().size();
if(size > 0){
printEntry(batchId,message.getEntries());
}
// 提交确认
connector.ack(batchId);
// 处理失败, 回滚数据
// connector.rollback(batchId);
}
} finally {
connector.disconnect();
}
}
private static void printEntry(long batchId,List<Entry> entrys) {
for (Entry entry : entrys) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue;
}
//数据反序列化
RowChange rowChage = RowChange.parseFrom(entry.getStoreValue());
EventType eventType = rowChage.getEventType();
for (RowData rowData : rowChage.getRowDatasList()) {
if (eventType == EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
3.2.2 连接canal server
private InetSocketAddress doConnect() throws CanalClientException {
try {
channel = SocketChannel.open();
channel.socket().setSoTimeout(soTimeout);
SocketAddress address = getAddress();
if (address == null) {
address = getNextAddress();
}
//1.连接 server
channel.connect(address);
//读channel
readableChannel = Channels.newChannel(channel.socket().getInputStream());
//写channel
writableChannel = Channels.newChannel(channel.socket().getOutputStream());
//2.握手
Packet p = Packet.parseFrom(readNextPacket());
if (p.getVersion() != 1) {
throw new CanalClientException("unsupported version at this client.");
}
if (p.getType() != PacketType.HANDSHAKE) {
throw new CanalClientException("expect handshake but found other type.");
}
Handshake handshake = Handshake.parseFrom(p.getBody());
supportedCompressions.add(handshake.getSupportedCompressions());
//
ByteString seed = handshake.getSeeds(); // seed for auth
String newPasswd = password;
if (password != null) {
// encode passwd
newPasswd = SecurityUtil.byte2HexStr(SecurityUtil.scramble411(password.getBytes(), seed.toByteArray()));
}
ClientAuth ca = ClientAuth.newBuilder()
.setUsername(username != null ? username : "")
.setPassword(ByteString.copyFromUtf8(newPasswd != null ? newPasswd : ""))
.setNetReadTimeout(idleTimeout)
.setNetWriteTimeout(idleTimeout)
.build();
//3.鉴权客户端
writeWithHeader(Packet.newBuilder()
.setType(PacketType.CLIENTAUTHENTICATION)
.setBody(ca.toByteString())
.build()
.toByteArray());
//4.连接成确认
Packet ack = Packet.parseFrom(readNextPacket());
if (ack.getType() != PacketType.ACK) {
throw new CanalClientException("unexpected packet type when ack is expected");
}
Ack ackBody = Ack.parseFrom(ack.getBody());
if (ackBody.getErrorCode() > 0) {
throw new CanalClientException("something goes wrong when doing authentication: "
+ ackBody.getErrorMessage());
}
connected = true;
return new InetSocketAddress(channel.socket().getLocalAddress(), channel.socket().getLocalPort());
} catch (IOException | NoSuchAlgorithmException e) {
throw new CanalClientException(e);
}
}