本文将介绍一套基于MySQL Binlog日志的同步系统。业务方基于这套系统，通过简单的配置部署即可将MySQL中的数据实时同步到其他数据库（例如Redis、ElasticSearch、HBase），实现数据的异构存储。

1. MySync诞生的背景

MySQL是我们日常业务开发中不可或缺的关系型数据库。很多时候，我们的业务因为读多写少的场景和各种需求，希望将MySQL中的数据同步到各种存储中。

企点客户库(https://qidian.qq.com/module/service.html)是企点客服产品中保存客户资源的核心模块。在企点客户库中，我们既需要提供高频低延时的客户资料查询接口，也需要支持对客户资料全文快速检索的功能；这些能力都不是MySQL的优势所在，用Redis来响应高频的查询请求，用ElasticSearch来支持客户资料全文索引的能力，显然是一种更好的选择。

如果需要在业务逻辑中实现对多个存储的写入，以下的情况会让很多开发者觉得非常头疼：在每个DML的地方，都需要维护对其他存储的增删改逻辑；如果MySQL的库表中新增或修改字段，写入其他存储的逻辑也要做相应的改造；如果有任意一个存储写入异常，不同存储中的数据就将不再一致，特别是在分布式的场景下（例如将MySQL中的数据同时同步到ES和Redis两种不同的存储来应对不同的业务需求，或者同步到多个Redis的实例来做存储容灾的备份），要保证不同存储实例中的数据完全一致的难度是比较大的。

如果有一套通用化的组件，不需要定制化的开发，通过简单的配置部署即可将MySQL中的数据实时同步到其他存储，必将在很大程度上提高各个业务方的开发效率；于是，企点MySQL同步组件——MySync应运而生。

2. MySync的架构

宏观上看，业务通过配置MySync，往DB中写入数据，就可以在配置的cache中直接读到DB中的数据。具体来看，应用将数据写入MySQL后，会产生Binlog数据，通过Binlog Reader的实时读取，写入消息队列中；各种存储写入组件，比如Redis Writer，从消息队列实时消费数据，写入其他异构存储，供读应用读取。Dump Tool可以用来做业务初始上线时的数据刷新导入。当然在业务允许的情况下，也可以通过更新DB全量数据的特定字段来刷出Binlog进行同步。

与许许多多优秀的开源项目一样，MySync中的模块是抽象化的，使用者可以根据自己的需求选择相应的模块。这些模块包括：通用的配置、日志、上报、消息模块；供Reader使用的检查点、高可用、消息生产模块；还有供Writer使用的消息消费、存储写入模块。

1. Reader组件简介

在Binlog中，我们主要关注的是两个event：一是Table_map_event，这里面包含了MySQL库表的schema信息；另一个是Row_event（注意，我们这里要求的Binlog format必须是row模式）这里面记录了每一行记录变更前后的详细信息。通过Binlog中的Table_map_event和Row_event，我们可以生成用于表示每一行记录具体变更的Base msg，它们在消息模块(ModMessage)序列化之后会发通过生产者模块(ModProducer)发给消息队列。

检查点模块(ModCheckpoint)用于记录Reader组件最近读取Binlog的位置。当我们启动Reader组件时，Reader会自动查找最近一次记录已读取Binlog的位置，并从此处开始读取Binlog；首次启动Reader组件时的Binlog位置需要通过配置文件指定。

高可用模块(ModHA)作为一种可选的配置，主要用于Reader实例的选主，避免Reader实例的单点问题，提高Reader服务的可用性。

消息生产模块(ModProducer)用于将序列化后的消息发送到消息队列，它需要实现消息到消息队列exactly once的语义（不重复、不丢失），来保证MySQL与不同Cache中数据的强一致性。

2. Writer组件简介

Writer组件从消息队列通过消费者模块(ModConsumer)获取消息，通过消息模块(ModMessage)把这些消息反序列化成标准化的Base msg；这些消息会通过消息的Hash Key路由到对应的存储写入模块(ModWriter)，最后由写入模块(ModWriter)来完成对Cache的写操作。

存储写入模块(ModWriter)会按照使用者的配置，生成对Cache的写入请求；这些写入请求必须严格有序执行，并保证执行成功，以此来保证MySQL与不同Cache中数据的一致性。目前的ModWriter已经有对ElasticSearch、Redis等写入的实现模块，未来将对更多的存储进行支持。

3. MySync的消息队列选型

可以看出，MySync同步的核心就是消息队列，那么消息队列的选型就至关重要。首先，我们需要消息队列有高可用性和可扩展性。更具挑战的一点是，我们在保证消息队列生产和消费高性能的前提下，最好能实现exactly once的语义。所谓exactly once，就是消息在生产消费的过程中，都有且只有一次，能做到不丢失、不重复；或者能通过其他的一些机制，能够解决消息乱序或者丢消息的问题，来保证MySQL和Cache数据的最终一致性。

目前，MySync在Kafka上有比较好的实践。

我们知道，当Binlog使用row模式时，每一条Binlog消息都代表了某个数据库表中一行完整的记录。如果我们指定消息中的Key，则可以很好的使用Kafka的Log Compaction功能，对于频繁变更的数据，只保存最后一条数据即可。这样消息队列中的历史数据无需删除，消息队列中数据的条数和数据库中的数据条数大体相当，使得消息队列数据可以永久保存，新接入的业务无需重新刷数据初始化，只需要从头消费即可，LinkedIn的CDC系统选择了Kafka作为消息中间件，也是基于这个原因。另一方面，Kafka的生产和消费均采用异步模式，使用了一些诸如打包发送、零拷贝、批量顺序写磁盘等手段，性能很高。与此同时，Kafka能很方便地支持分区的平行扩展，为扩容同步性能提供了可能。

那么我们如何通过Kafka来确保数据的一致性呢？这就需要保证消息在发送过程中不丢失、不重复、不乱序。

1. 生产

一种最简单的方式，就是将Kafka生产时的max.in.flight参数设为1，这个参数表示每个kafka连接最大的在途请求数，是一个量化异步发送消息的值：如果这个值是1，它必须发送一条消息，收到ack后，才能发送下一条。这样一来，exactly once是必然是实现了，但是性能可想而知会非常差。那如果这个值是不是1呢？这就意味着我们可以异步地批量发送消息，但这可能造成消息的乱序。比如我们异步地发送序列为1-10,11-20这两批消息，有没有可能是11-20这批消息先收到ack呢？完全有可能。那么我们怎么解决乱序问题呢？

其实，mysync只需要保证消息最终一致性即可。首先，我们会用一个本地的buffer作为缓冲区，将读到的Binlog都按序赋予一个序列号，放到这个buffer中，由专门的线程负责Binlog消息的发送、重传、确认。例如，我们现在有即将发送的序列号为1,2,3,4,5,6,7,8这8条消息，我们先发了1,2,3,4这4条，cur seq指针指向即将发送下一条Binlog消息的序列号，它目前会指向5。

由于这一过程是异步的，我们能收到消息ack回调的顺序未必是1,2,3,4。假如此时我们收到的ack是1,2,4，这时候就乱序了。

序列号为1,2的两条消息顺序并没有问题；那么，MySync就会把cur seq这个指针往回拨，因为我们在收到1和2的ack后之后，期望收到的ack是3，所以拨回到3，然后将3,4重新发一遍。

那么当序列号为3,4的两条消息都按序ack后，我们就可以确认1,2,3,4这四条消息都已经按序成功送达Kafka了。后续的消息将按照同样的确认逻辑依次处理

2. 消费

再来看Binlog消息的消费方。首先，所有的写入模块(ModWriter)必须严格按序执行每条Binlog消息的写入，并定期向Kafka提交已经执行成功Binlog消息对应消费组的offset；如果在此过程中有失败，则必须一直重试。如果重试不成功，则当前写入模块对应消费组的offset就不能提交；随着时间的流逝，Kafka的相关监控会发现在当前的topic中，已生产消息的offset和消费组已提交的offset的差距越来越大，会触发消费滞后的告警；此时业务方需要确认Cache写入速度异常的原因，及时做出相应的调整。

那么，生产方因网络丢包等原因出现消息乱序的问题，是否会对消费方造成影响呢？下面的这一串消息的序列号是我们之前在生产示例中出现乱序重传后，消费方收到Binlog消息序列的情况。可以看到，序列号为4的这条消息出现了2次，消费方可以很容易地通过序列号的异常发现消息重复、丢包的问题。同时，我们可以发现，即使Binlog消息有乱序的情况，只要最终能出现正确的序列，其写入Cache的最终结果是完全相同的，完全可以保证数据的最终一致性

3. Kafka的相关参数配置

最后我们分别从Kafka集群、生产、消费的配置方面来看看数据一致性的保证

1.生产（request.required.acks=-1）

Kafka producer对消息写入的ack有3种机制，如果request.required.acks设置成1，那么producer发送数据到leader，leader写本地日志成功，即返回客户端成功；此时ISR(In-Sync Replicas, 副本同步队列)中的副本还没有来得及拉取该消息，leader就宕机了，那么此次发送的消息就会丢失。request.required.acks设置成-1，意味着producer必须在follower副本确认接收到数据后才算一次发送完成，这就能保证了消息的持久性。

2.消费（enable.auto.commit=false; auto.offset.reset= earliest;）

Kafka的消费者默认会定期自动提交已消费消息的offset位置，但在MySync中，消费者必须自己管理offset，确保未被处理的Binlog消息不会被提交。如果MySync作为消费者拿不到上次已提交的offset，则直接从最早的位置开始消费数据。

3.broker（unclean.leader.election.enable=false、min.insync.replicas=2）

min.insync.replicas这个参数是配合request.required.acks使用的，它设定ISR中的最小副本数是多少，默认是1。对于我们每个partition三副本的配置，这里我们会将min.insync.replicas设为2，也就是说，除了每个partition的leader外，至少要有1个副本同步完成才算成功。

unclean.leader.election.enable参数的值为false，就意味着非ISR中的副本不能够参与partiton的选主。

结合CAP原理，MySync选择了CP without A，即如果不要求绝对的A（可用），则每个请求都需要在不同的Server之间保证C（强一致），而P（分区容错）会导致同步时间无限延长，如此CP也是可以保证的。

4. 分布式MySQL主备切换后如何寻找正确的Binlog位置

为了保证线上MySQL服务的高可用，我们会选择物理上一主一备的分布式MySQL实例；一条Binlog在不同的机器上可能位于不同Binlog文件中不同的位置，当MySQL发生主备切换时，读取Binlog的位置也会发生变化，那么MySync是如何在新的主机上找到正确的Binlog位置呢？

GTID(Global Transaction Identifier)是MySQL 5.6的新特性，它用于在Binlog中唯一标识一个事务，从而简化了MySQL的主备切换以及Failover的过程。MySQL Server在提交事务写Binlog时，会先写一条类型为GTID_Event的Binlog，用来指定下一个事务的GTID，之后再写事务的Binlog。主从同步时GTID_Event和事务的Binlog都会传递到从库，从库在执行的时候也是用同样的GTID写binlog，这样主从同步以后，就可通过GTID确定从库同步到的位置了。这样，我们可以直接通过GTID来寻找MySQL主备切换后已读取Binlog的位置，继续读取新的Binlog。

然而，在生产环境中MySQL实例的版本可能相较5.6更早，所以MySync面临的挑战是，在非GTID模式下完成MySQL主备切换后的自动同步。

MySync会在MySQL实例上构建一张表__db_mysync.t_binlog_record，并每分钟将这张表中唯一的记录值+1来构造锚点。如下图，在一段时间内__db_mysync.t_binlog_record的记录的值从10更新为11，在这期间产生了7条Row_event的Binlog，其中__db_mysync.t_binlog_record相关的Row_event即作为锚点。可以看到相同的Row_event在不同的机器上位于了不同Binlog文件的不同位置上，但是我们可以根据这些锚点来寻找在新旧主机上Binlog对应的位置。例如，MySync在读取mysql-bin.000052的position为200248的Binlog时，MySQL发生了主备切换，我们根据__db_mysync.t_binlog_record中记录变更为11的锚点，在新的MySQL主机上遍历最新的Binlog文件，查找到锚点记录为11的Binlog位于mysql-bin.000008的position为576204位置（如果查找不到，则继续遍历第二、第三、……、第N新的Binlog文件，直到找到锚点为止），并从这里开始重新读取Binlog。

由此，MySync完成了MySQL主备切换后寻找正确的Binlog位置的过程。

5. MySync的使用现状

MySync是腾讯开源协作奖获奖项目。目前企点已有MySync的多个实例在线上环境运行，服务了企点数万个MySQL库表同步到不同Cache的需求，自上线以来已稳定运行了5年，数据一致性已经得到了充分验证，并在很大程度上减少了业务在双写场景的开发工作量。MySync不仅在企点有广泛的使用，在公司内也越来越有影响力，目前已面向所有部门提供部署、使用的技术支持。

企点客户库模块是使用MySync的经典案例之一。客户库的原始数据保存在MySQL中；对于高频的客户基本字段查询请求，我们将MySQL的数据同步到Redis来完成；对于客户业务字段全文搜索的需求，我们将MySQL的数据同步到ES来完成；同时，我们可以直接监听Binlog Reader推送到Kafka的事件，为下游业务提供客户资料变更事件的实时推送。

E N D