使用Kafka订阅数据库的实时Binlog

微信扫一扫,分享到朋友圈

使用Kafka订阅数据库的实时Binlog

订阅 Binlog
的目的在于,实现实时的缓存更新、处理复杂逻辑数据实时同步到 Elasticsearch
/其它库-表等业务场景。

本篇内容包括:

  • 一种在应用层实现监听 SQL
    的方式
  • 预备知识:关于 Mysql
    事务的两阶段提交与 Binlog
  • 预备知识:关于 Kafka
  • 阿里云数据传输服务 DTS
    -数据订阅
  • 官方 DEMO
    的消费模型:生产->消费模型
  • 官方 DEMO
    提供的 MetaStore
    Checkpoint
    特性
  • 使用官方 DEMO
    需要注意的地方
  • 关于 avro
    序列化与反序列化

一种在应用层实现监听SQL的方式

笔者之前写过关于在应用中利用 Mybatis
插件和 SQL
解析工具实现监听 SQL
从而更新数据的文章,并且将这一功能整合到了个人的开源项目( easymulti-datasource-spring-boot-starter
)中,支持监听事务,支持实时消费监听到的 SQL
,也可以通过给事务监听器注册回调接口方式,在事务提交时才开始消费监听到的 SQL

使用 easymulti-datasource-spring-boot-starter
也能在应用层面轻松实现实时 SQL
订阅,但这种方式也存在弊端,虽然监听到 SQL
后也是异步消费,但拦截 SQL
、分析 SQL
,本身也会有点性能损耗。而如果是有多个应用修改同一个表的情况,那么就需要每个应用都写一遍消费的代码。

如果可以在更底层,直接订阅 Mysql
Binlog
,效率会比在应用层实现高得多。

预备知识:关于Mysql事务的两阶段提交与Binlog

Binlog
用于记录数据库执行的写入操作信息,以二进制的形式保存在磁盘中。

Binlog
Mysql
的逻辑日志,并且由 Server
层进行记录,无论使用何种存储引擎, Mysql
数据库都会记录 Binlog
日志。

Mysql
只有在事务提交时才会记录 Biglog
,并且事务在提交时, Biglog
还只是记录在内存中,然后才通过配置的刷盘策略写入到文件中。

Mysql
通过 sync_binlog
参数控制 Biglog
的刷盘时机,取值范围是 0-N

  • 0
    :由系统自行判断何时写入磁盘;
  • 1
    :每次 commit
    都将 Binlog
    写入磁盘;
  • N
    :每 N
    个事务 commit
    才将 Binlog
    写入磁盘;

毫无疑问, sync_binlog
最安全的是设置是 1
,这也是 MySQL 5.7.7
之后版本的默认值。

通常我们提到的 Mysql
事务的两阶段提交都与 InnoDB
存储引擎有关。

Mysql
事务分两个阶段提交,第一阶段由存储引擎预写记录,如 InnoDB
存储引擎写 Redolog
,此阶段 Binlog
不作任何操作;第二阶段首先是写 Binlog
,然后再由存储引擎完成事务的提交工作,如写入 commit
日记、释放锁等。

当第二阶段的写 Binlog
成功后, MySQL
就会认为事务已经提交并且持久化了,所以在这一步 Binlog
就已经可以发送给订阅者了。如果在写完 Binlog
后,存储引擎还没有完成提交的事务,刚好在这个时刻数据库崩溃,那么重启后依然能根据 Binlog
正确恢复该事务。如果在写 Binlog
这一步完成之前,任何操作的失败都会引起事务回滚。

所以,如果是直接订阅 Binlog
,我们并不需要关心事务最终是提交了还是回滚了,在事务未提交之前,我们都订阅不到该事务中执行的任何 SQL
的日记。

想要了解更多,推荐阅读文章:
MySQL
· 原理介绍 · 再议 MySQL
的故障恢复》

http://mysql.taobao.org/monthly/2018/12/04/

预备知识:关于Kafka

数据存储问题

Kafka
集群保留所有发布的记录,无论它们是否已被消费,可通过配置保留期限参数来控制消息的保留时长。如果保留策略设置为 2
天,一条记录发布后两天内,可以随时被消费,两天过后这条记录会被抛弃并释放磁盘空间。

offset消费偏移量

偏移量由消费者所控制,由消费者在消费记录后 commit
一个新的偏移量, kafka
会为消费者存储这个偏移量,以便于后续继续消费, kafka
会按 group + topic + partition
存储偏移量。当然,也可以自行存储,关于自行存储偏移量需要注意的问题后续会提到。

由于 kafka
group + topic + partition
存储偏移量,这同时也对应另一个问题:”同一个分组内,一个 topic
的每个 partition
都只能有一个消费者消费,但一个消费者可以同时消费多个 partition
。”

由于 offset
由消费者控制,所以消费者可以采用任何顺序来消费记录,也就是说,一个 topic
的任一消费者都可以重置到一个旧的偏移量,从而重新处理过去的数据,也可以跳过最近的记录,从当前位置开始消费。

消费者

一个 KafkaConsumer
实例并不一定就等于一个消费者。

subscribe
模式下,一个 KafkaConsumer
实例等于一个消费者。假设只有一个分区,开启多个 KafkaConsumer
,那么将会有一个消费者处于空闲状态,即这个线程每次调用该 KafkaConsumer
实例的 poll
方法都会一直返回空,拉不到任何消息,直到当前正在消费的 KafkaConsumer
长连接掉线后,重平衡后空闲的消费者才会拉取到记录。

这也证实了这句话: Kafka
实现消费的方式是将日志中的分区划分到每一个消费者实例上,以便在任何时间,每个消费者都是某个分区的唯一消费者。

subscribe
模式下,与其说一个 KafkaConsumer
等于一个消费者,不如说,一个连接( Socket
)等于一个消费者。

而在 assign
模式下,如果多个 KafkaConsumer
订阅的都是指定的 topic
和分区(并且同组),那么这些 KafkaConsumer
拉取的都会是同一个分区的记录。这里只是举例说明,不要这样用,否则会重复消费记录,两个线程交叉提交( commit
)偏移量( offset
)也会出问题。

消费者组

通常情况下,每个 topic
都会有一些消费组,一个消费组就是一个逻辑订阅者。

例如:

topic:用户注册
group 1:短信推送服务订阅者
group 2:邮件推送服务订阅者
复制代码

group1
group2
是逻辑订阅者,但每个逻辑订阅者下面都可以有多个消费者。

同一个组内的消费者数量不要超过 topic
partition
数量,因为超出 partition
数量的消费者不会被分配到 partition
,也就是会处于空闲状态(见”消费者”下的描述);

维护消费者组中的消费关系由 Kafka
协议动态处理,当有新的消费者加入组时,新加入的消费者将从组中其他成员处接管一些 partition
分区,当一个消费者消失时,该消费者拥有的分区将被重新分配给其它剩余的消费者。

还有一点,在同一个分组下,如果一个 topic
的每个分区当前都有一个消费者正在消费,新加入的消费者将会替代一个正在消费的消费者,接管被替代的消费者消费的分区。

阿里云数据传输服务DTS-数据订阅

阿里云数据传输服务 DTS
支持 MySQL
DRDS
Binlog
实时订阅。

我们可以不必使用官方提供的 SDK
订阅 Binlog
,而只需要使用 kafka
客户端,使用 kafka
API
实现 Binlog
订阅。

官方文档: 使用Kafka客户端消费订阅数据
https://help.aliyun.com/document_detail/121239.html?spm=a2c4g.11186623.6.785.6d4d6d2aIOqQQm

官方提供的 DEMO
subscribe_example
https://github.com/LioRoger/subscribe_example
,该 demo
龙玄
大佬提供。

我们选择基于官方 demo
[subscribe_example/javaimpl]构建 Mysql Binlog
实时订阅服务(试用阶段),而不是重复造轮子。但我们对源码做了部分修改,保留消息反序列化、 MetaStore
Checkpoint
特性。其中 MetaStore
Checkpoint
是这个 demo
最值得学习的地方。

官方DEMO的消费模型:生产->消费模型

DEMO
只开启一个消费者,这个消费者负责订阅消息,并将订阅到的消息放入一个阻塞队列( LinkedBlockingQueue
)中,这个阻塞队列的默认大小设置为 512

另外开启一个真正消费消息的线程,从该阻塞队列中读取消息并调用 RecordListener
consume
方法消费,在 RecordListener
消费完消息后,将该消息的 offset
包装成一个检查点( Checkpoint
),将该检查点设置为最新的检查点,另外会有一个定时任务每 5
秒提交一次最新的检查点,即提交 offset

kafka
消费者每次都有可能拉取到一批消息,并且这些消息是按发布顺序排好序的。因为 topic
的一个分区只能被一个消费者消费,而消息在分区中本就按消息的发布顺序排好序的。

DEMO
中,消费者将订阅到的消息放入阻塞队列也是按顺序放入,当队列满时会阻塞等待,因此只需要确保按顺序消费阻塞队列中的消息并提交 offset

如果不按顺序消费阻塞队列中的消息会怎样?

假设多个线程并行无顺序的消费拉取到的消息,那么就无法确保 offset
被正确提交,可能会导致部分消息重复消费。

在不严格要求每条消息都必须正确无异常地被消费的情况下,我们可以使用多线程消费,提升消息的消费速度。比如,消费阻塞队列中消息的线程只负责从阻塞队列获取消息,并负责解析,其它例如更新缓存等行为放到异步线程池中去执行,只要成功放入异步线程池,就更新 Checkpoint
offset
),继续消费后面的消息。

官方DEMO提供的MetaStore与Checkpoint特性

Checkpoint
用于记录分组内的一个 topic
的某个分区当前实际消费到的位置(偏移量: offset
)。

/**
* 安全检查点(即:记录消费位置)
*/
public class Checkpoint {
// 分区信息
private final TopicPartition topicPartition;
private final long timeStamp;
private final long offset;
public Checkpoint(TopicPartition topicPartition, long timeStamp, long offset) {
this.topicPartition = topicPartition;
this.timeStamp = timeStamp;
this.offset = offset;
}
}
复制代码

MetaStore
则用于存储 Checkpoint
,或者说是提交偏移量。

public interface MetaStore<V> {
Future<V> serializeTo(TopicPartition topicPartition, String group, V value);
V deserializeFrom(TopicPartition topicPartition, String group);
}
复制代码

DEMO
提供了两个实现类: KafkaMetaStore
LocalFileMetaStore
。其中 LocalFileMetaStore
实现的就是使用本地文件存储消费的分区的偏移量, KafkaMetaStore
则是调用 KafkaConsumer
commitAsync
方法异步提交偏移量,也就是说让 kafka
存储偏移量。

需要注意的是,在 subscribe
模式下,不要使用 LocalFileMetaStore
。当消费者以集群方式部署时,节点重启后由于 kafka
的再平衡,该节点消费的分区可能与重启之前的分区不同,那么本地文件存储的消费偏移量就使用不上,会导致从头(配置的初始化消费位置)开始消费记录;而如果只是部署一个消费者服务,或者多个消费者是在一个进程内的,又或是使用 assign
模式,那么可以使用 LocalFileMetaStore
,但需要确保每次服务重启都存在偏移量文件,如果切换服务器部署,则需要将偏移量文件同步到新的服务器上。

为了省去不必要的麻烦,我们直接弃用 LocalFileMetaStore
,而使用 KafkaMetaStore

public class KafkaMetaStore implements MetaStore<Checkpoint> {
private volatile KafkaConsumer kafkaConsumer;
//.....
// 异步提交offset
@Override
public Future<Checkpoint> serializeTo(TopicPartition topicPartition, String group, Checkpoint value) {
KafkaFutureImpl ret = new KafkaFutureImpl();
if (null != kafkaConsumer) {
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(value.getOffset(), String.valueOf(value.getTimeStamp()));
// 异步提交(不能同步提交,否则影响RecordGenerator#run())
// Notice: commitAsync is only put commit offset request to sending queue, the future  result will be driven by KafkaConsumer.poll() function
// So if you only call this method but not poll, you may not wait offset commit call back
kafkaConsumer.commitAsync(Collections.singletonMap(topicPartition, offsetAndMetadata), new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
if (null != exception) {
log.warn("KafkaMetaStore: Commit offset for group[" + group + "] topicPartition[" + topicPartition.toString() + "] " +
value.toString() + " failed cause " + exception.getMessage(), exception);
ret.completeExceptionally(exception);
} else {
log.debug("KafkaMetaStore:Commit offset success for group[{}] topicPartition [{}] {}", group, topicPartition, value);
ret.complete(value);
}
}
});
} else {
log.warn("KafkaMetaStore: kafka consumer not set, ignore report");
ret.complete(value);
}
return ret;
}
// 从kafka获取当前分区的offset和时间戳
@Override
public Checkpoint deserializeFrom(TopicPartition topicPartition, String group) {
if (null != kafkaConsumer) {
OffsetAndMetadata offsetAndMetadata = kafkaConsumer.committed(topicPartition);
if (null != offsetAndMetadata) {
return new Checkpoint(topicPartition, Long.valueOf(offsetAndMetadata.metadata()), offsetAndMetadata.offset(), offsetAndMetadata.metadata());
} else {
return null;
}
} else {
log.warn("KafkaMetaStore: kafka consumer not set, ignore fetch offset");
throw new KafkaException("KafkaMetaStore: kafka consumer not set, ignore fetch offset for group[" + group + "] and tp [" + topicPartition + "]");
}
}
}
复制代码

使用官方DEMO需要注意的地方

如果 topic
的某个分区从未被消费过,那么在首次启动消费者时,需要配置初始化消费位置,可以使用时间戳,也可以使用 offset
定位到想要消费的位置。

如果分区被消费过,那么就可以在消费者启动/重启时,先获取最后消费的位置,然后再从最后消费的位置开始消费。但由于 offset
是定时每 5
秒才提交一次,所以获取到的 offset
并不能代表实际消费的偏移量,所以每次重起都会有小部分记录被重新消费,这需要我们自行确保消息的幂等性消费。

官方 DEMO
默认使用 LocalFileMetaStore
,替换 MetaStore
只需要修改 RecordGenerator#getConsumerWrap
方法,代码如下:

public class RecordGenerator{
private ConsumerWrap getConsumerWrap(String message) {
// KafkaConsumer包装器
ConsumerWrap kafkaConsumerWrap = getConsumerWrap();
// 不建议使用LocalFileMetaStore存储(特别是部署到k8s上),否则将消费者部署到其它服务器后,需要将localCheckpointStore文件也要同步过去才可以
// metaStoreCenter.registerStore(LOCAL_FILE_STORE_NAME, new LocalFileMetaStore(LOCAL_FILE_STORE_NAME));
// 使用KafkaMetaStore
metaStoreCenter.registerStore(KAFKA_STORE_NAME, new KafkaMetaStore(kafkaConsumerWrap.getRawConsumer()));
// 从检查点存储器获取检查点(由于是每5秒提交一次,所以每次重起都会有小部分记录被重新消费)
Checkpoint checkpoint = getCheckpoint();
// 没有找到检查点,则使用配置的初始化检查点
if (null == checkpoint || Checkpoint.INVALID_STREAM_CHECKPOINT == checkpoint) {
checkpoint = initialCheckpoint; // 在配置文件中配置
log.info("RecordGenerator: use initial checkpoint [{}] to start", checkpoint);
} else {
log.info("RecordGenerator: load checkpoint from checkpoint store success, current checkpoint [{}]", checkpoint);
}
//.......
}
}
复制代码

最后,由于阿里云数据传输服务 DTS
-数据订阅只会将日记提交到一个分区,即一个 topic
只有一个分区,这是为了确保能够按正确的顺序消费每一条 sql
日记。所以,我们没有必要使用 subscribe
模式,应该使用 assign
模式。

DefaultConsumerWrap
封装了 KafkaConsumer
,使用 assign
模式在该类的 assignTopic
方法表现,代码如下。

public class DefaultConsumerWrap extends ConsumerWrap {
private KafkaConsumer<byte[], byte[]> consumer;
@Override
public void assignTopic(TopicPartition topicPartition, Checkpoint checkpoint) {
// KafkaConsumer
consumer.assign(Collections.singletonList(topicPartition));
log.info("RecordGenerator:  assigned for {} with checkpoint {}", topicPartition, checkpoint);
// 设置消费位置
setFetchOffsetByTimestamp(topicPartition, checkpoint);
}
}
复制代码

其中, assignTopic
方法的第二个参数( Checkpoint
)从 MetaStore
获取而来,或者是使用配置的初始化位置。在调用 KafkaConsumer#assign
方法之后,调用 setFetchOffsetByTimestamp
方法设置消费位置,后续就可以调用 KafkaConsumer#poll
方法拉取消息了。

setFetchOffsetByTimestamp
方法实现如下,相比 DEMO
源码,我们做了点修改。

public class DefaultConsumerWrap extends ConsumerWrap {
@Override
public void setFetchOffsetByOffset(TopicPartition topicPartition, Checkpoint checkpoint) {
// 移动到指定位置继续消费
consumer.seek(topicPartition, checkpoint.getOffset());
}
// recommended
@Override
public void setFetchOffsetByTimestamp(TopicPartition topicPartition, Checkpoint checkpoint) {
// 优先使用偏移量
if (checkpoint.getOffset() > 0) {
setFetchOffsetByOffset(topicPartition, checkpoint);
return;
}
long timeStamp = checkpoint.getTimeStamp();
// 根据时间戳获取偏移量
Map<TopicPartition, OffsetAndTimestamp> remoteOffset = consumer.offsetsForTimes(Collections.singletonMap(topicPartition, timeStamp));
OffsetAndTimestamp toSet = remoteOffset.get(topicPartition);
if (null == toSet) {
throw new RuntimeException("RecordGenerator:seek timestamp for topic [" + topicPartition + "] with timestamp [" + timeStamp + "] failed");
}
// 移动到指定位置继续消费
consumer.seek(topicPartition, toSet.offset());
}
}
复制代码

关于avro序列化与反序列化

官方提供的 demo
,其中 com.alibaba.dts.formats.avro
这个 package
是由 avro
shcema
编译而来的,我们也可以自行编译,具体实现如下:

1、执行命令编译 avsc
文件生成 java
代码

java -jar avro/avro-tools-1.8.2.jar compile -string schema avro/Record.avsc .
复制代码

2、将生成的 com.alibaba.dts.formats.avro
这个 package
拷贝到当前工程根目录下面,当然,也可以封装到一个模块,在主模块中引入。

nginx last 和 break 的区别

上一篇

PEP 257:Docstring书写规范

下一篇

你也可能喜欢

使用Kafka订阅数据库的实时Binlog

长按储存图像,分享给朋友