在微服务拆分的架构中，各服务领有本人的数据库，所以经常会遇到服务之间数据通信的问题。比方，B服务数据库的数据来源于A服务的数据库；A服务的数据有变更操作时，须要同步到B服务中。
第一种解决方案:
在代码逻辑中，有相干A服务数据写操作时，以调用接口的形式，调用B服务接口，B服务再将数据写到新的数据库中。这种形式看似简略，但其实“坑”很多。在A服务代码逻辑中会减少大量这种调用接口同步的代码，减少了我的项目代码的复杂度，当前会越来越难保护。并且，接口调用的形式并不是一个稳固的形式，没有重试机制，没有同步地位记录，接口调用失败了怎么解决，忽然的大量接口调用会产生的问题等，这些都要思考并且在业务中解决。这里会有不少工作量。想到这里，就将这个计划排除了。
第二种解决方案：
通过数据库的binlog进行同步。这种解决方案，与A服务是独立的，不会和A服务有代码上的耦合。能够间接TCP连贯进行传输数据，优于接口调用的形式。 这是一套成熟的生产解决方案，也有不少binlog同步的中间件工具，所以咱们关注的就是哪个工具可能更好的构建稳固、性能满足且易于高可用部署的计划。
通过调研，咱们抉择了canal[https://github.com/alibaba/canal]。canal是阿里巴巴 MySQL binlog 增量订阅&生产组件，曾经有在生产上实际的例子，并且不便的反对和其余罕用的中间件组件组合，比方kafka，elasticsearch等，也有了canal-go go语言的client库，满足咱们在go上的需要，其余具体内容参阅canal的github主页。
原理简图
OK，开始干！当初要将A数据库的数据变更同步到B数据库。依据wiki很快就用docker跑起了一台canal-server服务，间接用canal-go写canal-client代码逻辑。用canal-go间接连canal-server，canal-server和canal-client之间是Socket来进行通信的，传输协定是TCP，交互协定采纳的是 Google Protocol Buffer 3.0。
工作流程
1.Canal连贯到A数据库，模仿slave
2.canal-client与Canal建设连贯，并订阅对应的数据库表
3.A数据库产生变更写入到binlog，Canal向数据库发送dump申请，获取binlog并解析，发送解析后的数据给canal-client
4.canal-client收到数据，将数据同步到新的数据库
Protocol Buffer的序列化速度还是很快的。反序列化后失去的数据，是每一行的数据，依照字段名和字段的值的构造，放到一个数组中 代码简略示例:
func Handler(entry protocol.Entry) { var keys []string rowChange := &protocol.RowChange{} proto.Unmarshal(entry.GetStoreValue(), rowChange) if rowChange != nil { eventType := rowChange.GetEventType() for _, rowData := range rowChange.GetRowDatas() { // 遍历每一行数据 if eventType == protocol.EventType_DELETE || eventType == protocol.EventType_UPDATE { columns := rowData.GetBeforeColumns() // 失去更改前的所有字段属性 } else if eventType == protocol.EventType_INSERT { columns := rowData.GetAfterColumns() // 失去更后前的所有字段属性 } ...... } }} 遇到的问题
为了高可用和更高的性能，咱们会创立多个canal-client形成一个集群，来进行解析并同步到新的数据库。这里就呈现了一个比拟重要的问题，如何保障canal-client集群解析生产binlog的程序性呢？
咱们应用的binlog是row模式。每一个写操作都会产生一条binlog日志。 举个简略的例子：插入了一条a记录，并且立马批改a记录。这样会有两个音讯发送给canal-client，如果因为网络等起因，更新的音讯早于插入的音讯被解决了，还没有插入记录，更新操作的最初成果是失败的。
怎么办呢？canal能够和音讯队列组合呀!而且反对kafka，rabbitmq，rocketmq多种抉择，如此优良。咱们在音讯队列这层来实现音讯的程序性。(前面会说怎么做)
抉择canal+kafka计划
咱们抉择了音讯队列的业界标杆: kafka UCloud提供了kafka和rocketMQ音讯队列产品服务，应用它们可能疾速便捷的搭建起一套音讯队列零碎。减速开发，不便运维。
上面就让咱们来一探到底：
①抉择kafka音讯队列产品，并申请开明
②开明实现后，在治理界面，创立kafka集群，依据本身需要，抉择相应的硬件配置
③一个kafka+ZooKeeper集群就搭建进去了，给力！
并且蕴含了节点治理、Topic治理、Consumer Group治理，可能十分不便的间接在控制台对配置进行批改
监控视图方面，监控的数据包含kafka生成和生产QPS，集群监控，ZooKeeper的监控。可能比较完善的提供监控指标。
canal的kafka配置
canal配上kafka也十分的简略。 vi /usr/local/canal/conf/canal.properties
# ...# 可选项: tcp(默认), kafka, RocketMQcanal.serverMode = kafka# ...# kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092canal.mq.servers = 127.0.0.1:9002canal.mq.retries = 0# flagMessage模式下能够调大该值, 但不要超过MQ音讯体大小下限canal.mq.batchSize = 16384canal.mq.maxRequestSize = 1048576# flatMessage模式下请将该值改大, 倡议50-200canal.mq.lingerMs = 1canal.mq.bufferMemory = 33554432# Canal的batch size, 默认50K, 因为kafka最大音讯体限度请勿超过1M(900K以下)canal.mq.canalBatchSize = 50# Canal get数据的超时工夫, 单位: 毫秒, 空为不限超时canal.mq.canalGetTimeout = 100# 是否为flat json格局对象canal.mq.flatMessage = falsecanal.mq.compressionType = nonecanal.mq.acks = all# kafka音讯投递是否应用事务canal.mq.transaction = false# mq configcanal.mq.topic=default# dynamic topic route by schema or table regex#canal.mq.dynamicTopic=mytest1.user,mytest2..*,.*..*canal.mq.dynamicTopic=mydatabase.mytablecanal.mq.partition=0# hash partition configcanal.mq.partitionsNum=3canal.mq.partitionHash=mydatabase.mytable具体见：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
解决程序生产问题
看到上面这一行配置
canal.mq.partitionHash=mydatabase.mytable咱们配置了kafka的partitionHash，并且咱们一个Topic就是一个表。这样的成果就是，一个表的数据只会推到一个固定的partition中，而后再推给consumer进行生产解决，同步到新的数据库。通过这种形式，解决了之前碰到的binlog日志程序解决的问题。这样即便咱们部署了多个kafka consumer端，形成一个集群，这样consumer从一个partition生产音讯，就是生产解决同一个表的数据。这样对于一个表来说，就义掉了并行处理，不过集体感觉，凭借kafka的性能弱小的解决架构，咱们的业务在kafka这个节点产生瓶颈并不容易。并且咱们的业务目标不是实时一致性，在肯定提早下，两个数据库保障最终一致性。
下图是最终的同步架构，咱们在每一个服务节点都实现了集群化。全都跑在UCloud的UK8s服务上，保障了服务节点的高可用性。
canal也是集群换，然而某一时刻只会有一台canal在解决binlog，其余都是冗余服务。当这台canal服务挂了，其中一台冗余服务就会切换到工作状态。同样的，也是因为要保障binlog的程序读取，所以只能有一台canal在工作。
并且，咱们还用这套架构进行缓存生效的同步。咱们应用的缓存模式是:Cache-Aside。同样的，如果在代码中数据更改的中央进行缓存生效操作，会将代码变得复杂。所以，在上述架构的根底上，将简单的触发缓存生效的逻辑放到kafka-client端对立解决，达到肯定解耦的目标。
目前这套同步架构失常运行中，后续有遇到问题再持续更新。
更多内容，欢送点击下方作者主页进行交换~
本文作者：UCloud利用研发工程师 Cary