吴夏，腾讯云 TDSQL 研发工程师，目前次要负责日志解析复制、数据传输同步模块的开发工作。
一、场景及需要
在金融业务场景中，数据的同步、订阅、散发是常见需要，例如保险行业常见的总分零碎架构，多个子库须要实时地将业务数据同步至总库汇总查问；银行外围交易系统中，须要将交易数据实时同步至剖析子系统进行报表、跑批等业务操作。
因而，腾讯云打造的分布式数据库 TDSQL（Tencent distribute database）作为一个金融级数据库产品，对数据的散发、解耦能力是必不可少的。
▲ 基于总分的数据汇总架构
▲ 数据实时同步剖析架构
其中，TDSQL-MULTISRCSYNC（下文简称多源同步）模块正是为了应答这样的需要和场景所开发的高性能、高统一、反对多种异构数据平台的数据散发服务。
该服务反对以 TDSQL 作为源端的数据实时同步散发至 MySQL、Oracle、Postgres、Kafka 等平台，同时也反对以 TDSQL 作为指标端，将 MySQL 或者 Oracle 的数据实时同步至 TDSQL 中，并且部署灵便，反对一对多、多对一等多种复制拓扑构造。以后该服务的官网名称为“数据同步”，是作为一个子性能集成在腾讯云 TencentDB for TDSQL 产品中。
二、零碎架构
多源同步模块典型的基于日志的 CDC 复制技术，其零碎架构如下：
从上图咱们能够看到，整个零碎能够大抵分成三个局部：producter，store，consumer。
1、producter
增量日志获取模块，次要负责解析获取源端的增量数据改变日志，并将获取到的日志解析封装为 JSON 协定的音讯体，投送至 Kafka 音讯队列。
当源端是 MySQL 或者 TDSQL 时，获取的增量日志为 binlog 事件，这里要求 binlog 必须是 row 格局且为 full-image。当源端是 Oracle，producter 从 Oracle 的物化视图日志中获取增量数据并进行封装和投送。
这里 producter 在向 Kafka 生产音讯时，采纳 at-least-once 模式，即保障特定音讯队列中至多有一份，不排除在队列中有音讯反复的状况。
2、store
这里采纳 Kafka 作为两头存储队列，因为数据库系统日志有程序性要求，因而这里所有的 topic 的 partition 个数均为 1，保障可能按序生产。
3、consumer
日志生产和重放模块，负责从 Kafka 中将 CDC 音讯生产进去并依据配置重放到指标实例上。这里因为 producter 端采纳 at-least-once 模式生产，因而消费者这里实现了幂等逻辑保证数据重放的正确。
三、外围设计及实现
1、基于行的哈希并发策略
金融业务场景中，往往对数据的实时性要较高，因而对数据同步的性能提出了比拟高的要求。为了解决这样的问题，consumer 采纳了基于行的哈希并发策略实现并行重放。上面以 binlog 音讯为例来阐明该策略的实现。
MySQL 在记录 binlog 时，依照事务的提交程序将行的改变写入 binlog 文件，因而依照 binlog 文件记录事件的程序进行串行重放，源端和指标端数据库实例状态肯定会达到统一。
然而串行重放因为速度慢，在遇到如批量更新等大事务时，容易产生较大的同步时延，适应不了对数据实时性较高的同步场景。
为了进步并发度，consumer 依照每个行记录的表名和主键值进行 hash，依据 hash 值将音讯投送到对应的同步线程中。
有些读者在这里可能会有疑难，这样乱序的重放难道不会导致数据不统一吗？答案是不会的，因为尽管是将程序的音讯序列打乱了，然而同一行的所有操作都是在同一个线程中是有序的，因而只有每个行的改变执行序列正确，最终数据是会统一。
这个过程如下图所示：
目前，基于行级的并发单任务同步速率能够达到 4W 的 QPS，曾经能够满足绝大多数场景对同步速率的要求。
这里每个线程在重放的时候，都会将音讯依照肯定的数量封装成事务来进行重放。这种模式下的并发复制，实际上实现的是最终一致性，因为原有的事务构造曾经被突破。当然因为并发复制速度够快，业务如果可能承受秒级的同步时延，基本上业务是感知不到不统一的数据。
2、row 格局 binlog 事件的幂等容错
实现幂等逻辑的动机有三个：
因为生产者实现的是 at-least-once 模式进行音讯生产，因而 consumer 这里必须要是否解决音讯反复的问题。
反对幂等逻辑后，便于数据的修复，且在数据同步的过程中不须要记录镜像点，便于运维。
反对主动容错，升高同步失败，卡住的概率。
这里幂等逻辑的设计准则就是，保障依照 binlog 事件的用意去对指标实例进行批改且肯定要胜利。
如 insert 事件，其用意就是要在数据库中有一条 new 值标识的记录；update 事件的用意就是，数据库中没有 old 值标识的记录，只有 new 值标识的记录；delete 操作也是同样，其后果就是要求指标数据库中，不蕴含 old 值标识的记录。
因而针对 insert，update，delete 操作，其幂等逻辑如下：
1）INSERT
依据上图能够看到，当呈现主键抵触时，insert 操作会转变成 delete+insert 操作来保障 insert 动作执行胜利。另外图中的影响行数小于 0 或者等于 0 标识执行 SQL 出错和主键抵触。
2）update
从上图咱们能够看到，update 操作的幂等解决，其实就是保障了在数据库中，只能有 new 值产生的记录。
3）delete
从上图能够看到，delete 的幂等准则就是，确保指标 DB 中没有 delete 事件中标识的记录。
在实现了上述的幂等逻辑后，会带来很多便当。如在全量迁徙数据时，无需在记录镜像点，只有保障增量日志获取的工夫比全量镜像点早，即使有 binlolg 的重放，因为有幂等逻辑，也能保障最终的数据统一。
3、多惟一约束条件下的并发管制
从下面的原理图能够看出，在 Kafka 队列中，具备雷同主键值的记录会被投送到雷同的线程，且线程内是有序的。这样的并发形式在上面这样的场景中，会产生数据不统一的状况。以下是对该场景的详细描述。
表 sync-table 的表构造定义如下：
当初在源端 MySQL 执行下列操作序列：
该操作序列会产生三条 binlog，别离是 insert (1,lucy,18)，delete(1,lucy,18)，insert(2,lucy,20)。那么这三条 binlog 事件会依照下图所示的形式散发到不同的同步线程：
因为线程间的执行程序是齐全并发的，因而这三个操作在两个线程间的执行程序可能为以下几种状况。
1）线程 1 比线程 2 执行得早
指标实例这种执行程序与源实例的执行程序完全一致，不会造成数据的不统一。
2）线程 1 和线程 2 执行时序有重叠
当线程 2 执行 insert 时，因为在这之前线程 1 曾经将惟一索引为 lucy 的记录写入了 DB，因而线程 2 的操作会失败（惟一索引抵触），从而进入幂等流程。
这里 insert 的幂等逻辑是会依据记录中的惟一索引字段先进行一次删除操作，即执行 delete where id = 2 和 delete where name = ‘lucy’；而后再将记录插入到数据库，执行 insert(2,lucy,20)，保障插入的胜利。
之后线程 1 再执行删除操作时，也会进入幂等流程（因为（1，Lucy，8）不存在，delete 的影响行数为 0），最终目标实例的状态是存在记录（2，lucy，20）。后果正确，不会造成不统一。
3）线程 2 先与线程 1 执行结束
线程 2 执行完 insert 后，线程 1 执行 insert 会因为惟一索引束缚抵触而报错失败，从而进入幂等流程。
线程 1 会执行 delete where id = 2 和 delete where name = ‘lucy’；之后在执行 insert(1,lucy,18)。最终当线程 1 全副执行完后，指标实例内不存在（2，lucy，20）这条记录。造成了数据的不统一。
通过上述的问题形容，咱们能够发现，产生数据不统一的起因其实是以后的数据并发策略在多惟一束缚的条件下不能依照正确的时序来进行重放。
因而在解决这种既有主键又蕴含一个或多个惟一索引表的数据时，咱们就须要额定的伎俩来保障散布在多个线程中的 binlog 事件按序执行。
这里 consumer 采纳的解决方案是在散发 binlog 事件到多个同步线程中的时候，同时下发一个锁构造，来协调多个线程中含有雷同惟一束缚值 binlog 事件的执行程序。如下图所示：
这里线程 1 的 delete 事件与线程 2 的 insert 事件关联这一个锁构造，如果线程 1 的 delete 事件没有执行完结，则线程 2 的 insert 事件不会执行。
① 锁的设计
依据下面的剖析咱们晓得，当一个表的束缚定义除了蕴含主键外，还蕴含惟一索引的话，则须要保障雷同惟一索引的事件依照程序来执行。
因而这里锁的语义应该是，如果有前序蕴含雷同惟一索引值的事件没有执行完，则须要期待，待其执行完后再执行以后事件。consumer 的锁构造如下所示：
所有的锁构造是寄存在一个全局的数组中，在锁下发的时候，依据表名做一次 hash，失去数组的下标。数组中的每一项蕴含了一个 hash_map 构，其中 key 由表名 + 惟一索引列名 + 该列的值形成，类型为字符串；该 key 对应的 value 值为一个锁构造的指针 lock*。
lock 构造中蕴含下列成员：
② 锁的下发
当 consumer 的 dispatch 线程对音讯进行散发时，首先检测，该音讯所对应的表是否蕴含初主键外的惟一束缚，如果有的话，则须要在下发该条音讯时，一并下发锁构造。
开始时，会首先依据表名和惟一索引的信息，查问是否蕴含该锁构造，如果蕴含则间接进入下发流程，如果不蕴含，则创立一个锁，并将其写入 lock_map 中，而后开始锁下发：
自增锁构造中的 wait-count；
保留 event-id 到变量 wait_event_id，之后将本人的 event_id 写入锁构造中；
更新 cond_map, 将本人 event_id 的键值对写入 cond_map , 并标识为持有该锁；
将该锁构造句柄和在第二步保留下的 event_id，随着音讯体一并下发到同步线程。
③ 锁的保护
同步线程在解决音讯时，首先会检测改音讯体是否蕴含锁构造，如果蕴含锁构造的话，解决流程如下所示：
check 锁构造中的 cond_map，查看 key 为 wait_event_id 的锁是否开释，如果没有开释则开始 pthread_cond_wait()。
当收到条件变量告诉时，检测 cond_map 中 wait_event_id 的锁是否开释，如果没有开释则持续 wait()。
当收到条件变量告诉时，检测到 cond_map 中 wait_event_id 的锁曾经开释，则开始对该音讯进行重放。
重放该音讯完结后，更新锁构造中的 wait-count 减 1。如果更新后 wait-count 等于 0，则阐明该锁上曾经没有任何音讯持有。销毁改锁。
如果更新后 wait-count 大于 0，则阐明还有音讯再期待该锁构造。更新 cond_map，将本人 event_id 对应的 value 更新为开释状态，并且将 wait_event_id 对应的键值对删除。
执行完上述操作后执行 broadcast() 操作，告诉其余期待线程。
4、优化与瞻望
目前，多源同步已在腾讯私有云、专有云等多家金融客户业务中运行，提供牢靠的数据散发同步能力。后续会在异构平台接入等能力上做更多的投入，如 DB2、SQLserver、大数据平台等，以适应更多业务场景。