概述canal是阿里巴巴旗下的开源项目，纯Java开发。基于数据库增量日志分析，提供增量数据订阅消费。目前主要支持MySQL（也支持mariaDB）。背景早期阿里巴巴B2B公司因杭州和美国双机房部署，有跨机房同步的业务需求。但是早期的数据库同步业务主要是通过触发器获取增量变化。但从2010年开始，阿里巴巴各公司开始逐步尝试基于数据库日志分析获取增量变化进行同步，进而衍生出增量变化。大众订阅消费业务从此开启了一个新时代。附言。目前内部使用的同步已经支持mysql5.x和oracle部分版本的日志分析。基于日志增量订阅消费支持业务：数据库镜像数据库实时备份多级索引（买卖双方各自分库索引）搜索构建业务缓存刷新价格变动等重要业务消息目前渠道支持源端MySQL5.1.x、5.5.x、5.6.x、5.7.x、8.0.x工作原理语句执行消耗的时间。主要用于备份和数据同步。binlog有三种模式：STATEMENT、ROW、MIXEDSTATEMENT记录执行的sql语句ROW记录真实的行数据记录MIXED记录1+2，按照1的模式记录。比如下面的sqlCOPYupdateusersetage=20对应在STATEMENT模式下只有一条记录，而在ROW模式下可能有数千条记录（取决于数据库中的记录数）。MySQL主从复制原理Slave上的IO线程连接Master，从指定日志文件的指定位置（或从初始日志）请求日志内容；Master收到Slave的IO线程的请求后，负责复制的IO线程根据请求信息读取指定日志指定位置后的日志信息，返回给Slave上的IO线程边。返回的信息除了日志中包含的信息外，还包括Master端的BinaryLog文件名和本次返回的信息在BinaryLog中的位置；Slave的IO线程接收到信息后，将收到的日志内容依次写入到Slave端的RelayLog文件（mysql-relay-bin.xxxxxx）的末尾，记录下日志的文件名和位置将Master端的bin-log读取到master-info文件中，以备下次读取时清楚，高速Master》需要从某个bin-log开始的日志内容请发给我”Slave的SQL线程检测到RelayLog中有新的内容后，会立即对内容进行解析。Log文件中的内容在Master端真正执行的时候就变成了可执行的Query语句，自己去执行这些Query语句。这样实际上在Master端和Slave端都执行了同一个Query，所以两端的数据是完全一样的。当然，这个过程还是有一定延迟的。mysql的binlog文件是这样的。COPYmysql-bin.003831mysql-bin.003840mysql-bin.003849mysql-bin.003858开启Binlog时需要注意以下几点：Master主库一般有多个Slave订阅，Master主库必须支持实时改变业务系统的操作。资源会出现瓶颈；需要同步的数据表必须有主键；canal可以同步数据的原理了解了mysql的主从同步机制，再看canal就更清楚了。主服务器拉取数据。canal模拟mysqlslave的交互协议，伪装成mysqlslave，向mysqlmaster发送dump协议。mysqlmaster收到dump请求，开始pushbinarylog给slave（也就是canal）。canal解析二进制日志对象（本来就是字节流）canalcanal架构的设计理念canal的组件化设计非常好，有点类似于tomcat的设计。使用组合设计、依赖倒置和面向接口的设计。canal的组件，canalserver，代表我们部署的一个canal应用。Canal实例表示一个canal服务器中的多个mysql实例。从这一点来看，说明一个canalserver可以从多个库中采集数据，在canal中称为destionation。每个运河实例由多个组件组成。这些组件在conf/spring/default-instance.xml中配置。他其实是用spring容器来管理这些组件的。实例中包含的组件以下是cannalInstance作业中包含的大型组件。从conf/spring/default-instance.xmlCOPYEventParser设计eventParser最基本的组件，类似于mysql从库的dump线程，负责从master获取bin_log。整个parser过程大致可以分为几个步骤：Connection获取上次解析成功的位置（如果是第一次启动，会获取初始指定的位置或者当前数据库的binlog位置）Connection建立链接，发送BINLOG_DUMP指令//0.writecommandnumber//1.写入4字节的bin-logpositiontostartat//2.写入2字节的bin-logflags//3.写入4字节的slaveserverid//4.写入bin-logfilenameMysql开始推送接收到的BinalyLog使用Binlogparser解析协议，补充一些具体信息//补充字段名，字段类型，主键信息，unsigned类型处理，传递给EventSink模块进行数据存储。日志定位EventSink设计eventSink数据采集，使用设置的filter对binlog进行过滤，工作过程如下：数据过滤：支持通配符过滤方式，表名，字段内容等数据路由/分发：解决1:n(一个parser对应多个store的模式）数据合并：解决n:1（多个parser对应一个store）数据处理：在进入store之前进行额外的处理，比如join数据1:n业务合理使用数据库资源，一般常见的业务按照schema进行隔离，然后在上层的mysql或者dao层面做一个数据源路由，屏蔽数据库物理位置对开发的影响。阿里系统主要通过cobar/tddl解决数据源路由问题。因此，一般情况下，一个数据库实例上会部署多个schema，每个schema都会有一个或多个业务方关注数据n:1的业务。同样，当一个业务的数据规模达到一定程度时，必然会涉及到水平拆分和垂直拆分。当需要处理拆分后的数据时，需要联动多个store进行处理，消费站点会变成多份，数据消费的进度无法完全实现。可能的订购保证。因此，在某些业务场景下，需要对拆分后的增量数据进行归并，比如根据时间戳/全局id进行排序归并。EventStore旨在存储filter过滤后的数据，目前canal的数据只存储在这里，工作流程如下目前只实现了Memory内存模式，计划增加本地文件存储将来。mixed混合模式借鉴了Disruptor的RingBuffer的实现思路，定义了3个cursorPut：Sink模块的最后写入位置，用于数据存储。Get:数据订阅获取的最后一个提取位置Ack:数据消费成功的最后一个消费位置参考Disruptor的RingBuffer的实现，理顺RingBuffer：实现描述：Put/Get/Ack游标用于递增，使用long类型存储缓冲区的获取操作是通过取余或与操作来执行的。（并且操作：cusor&(size-1)，size需要是2的索引，效率比较高）metaManagermetaManager用来存储一些原始数据，比如消费的游标，当前活跃的服务器等信息alarmHandleralarmHandler报警，一般是这样就是错误日志。理论上应该可以自定义邮件等形式，但是目前不支持各个组件目前支持的类型。canal使用springbean容器组装一个canal实例，目的是更加灵活。Canal可以通过这些组件的选择来达到不同使用场景的效果。例如，如果是单机，一般使用文件来存储元数据。对于HA，一般使用zookeeper来存储元数据。eventParsereventParser目前解析mysql日志的MysqlEventParser只有三种GroupEventParser多个eventParser的集合，理论上对应分表的情况，可以通过这个合并在一起RdsLocalBinlogEventParser是基于rdsbinlog的复制eventSinkeventSink目前只有EntryEventSink是基于mysql对binlog数据对象eventStoreeventStore的处理操作，目前只有一种类型的MemoryEventStoreWithBuffer，其内部使用了一个ringbuffer，也就是说canal解析出来的数据是存储在内存中，而不是zookeeper中。有许多metaManager元管理器。其实按照元数据存储的位置可以分为三类：内存，文件，zookeeperCanal-HA机制canal支持HA，其实现机制也是zookeeper实现的。使用的特性是watcher和EPHEMERAL节点（绑定session生命周期），类似于HDFSHA。Canal的HA分为两部分。Canalserver和canalclient都有对应的HA实现。canalserver：为了减少对mysqldump的请求，不同服务器上的实例（不同服务器上的同一个实例）每次只需要一个实例在运行，其他的都处于standby状态（standby是instance的状态）.canalclient：为了保证有序性，同一时间只能有一个canalclient对一个instance进行get/ack/rollback操作，否则无法保证client收到的顺序。serverha的架构图如下：当canalserver要启动一个canal实例时，首先对zookeeper_做一个试启动判断（实现：创建一个EPHEMERAL节点，谁创建成功谁允许启动）zookeeper节点创建成功后，对应的canalserver启动对应的canal实例，未成功创建的canal实例会处于standby状态。一旦zookeeper发现canalserverA创建的实例节点消失了，它会立即通知其他canalserver重新执行步骤1，重新选择一个canalserver来启动实例。每次canal客户端连接时，都会先询问当前正在启动canal实例的zookeeper，然后与其建立链接。一旦链接不可用，它将尝试重新连接。CanalClient方式与CanalServer方式类似，也是通过zookeeper抢占EPHEMERAL节点进行控制。Canal工作进程dumplog启动后，去MySQL执行dump操作的binlog位置，确定工作进程。当启动一个canal实例时，首先启动一个eventParser线程来转储数据。他去找master拉binlog的时候，需要binlog所在的位置。位置按以下顺序确定（这个地方是关于HA模式的ha）。启动时判断是否使用zookeeper。如果是zookeeper，看能不能拿到cursor（也就是binlog信息）。如果能拿到，把这个信息存入内存（MemoryLogPositionManager），然后把这个信息拿给mysql。如果通过1获取不到dumpbinlog（一般是zookeeper的各个，比如第一次搭建的时候，或者zk中的数据因为某些原因被删除了），去配置文件配置获取，把这个信息存入内存（MemoryLogPositionManager），然后把这个信息拿到mysql中的dumpbinlog中。如果2还没有搞定，就到mysql中执行一条sqlshowmasterstatus。这条语句会显示当前mysqlbinlog的最后一个位置的信息，也就是刚刚写入的binlog的位置信息。将这些信息存储在内存（MemoryLogPositionManager）中，然后将这些信息dump到mysql中的binlog。eventParser后续的操作会使用内存中保存的binlog位置（MemoryLogPositionManager）去master上进行dump操作。mysqlshowmasterstatusoperationCOPYmysql>showmasterstatus\G******************************1.行**************************文件：mysql-bin.000028位置：635762367Binlog_Do_DB：Binlog_Ignore_DB：Executed_Gtid_Set：18db0532-6a08-11e8-a13e-52540042a113:1-2784514,318556ef-4e47-11e6-81b6-52540097a9a8:1-30002,ac5a3780-63ad-11e8-a9ac-52540042a113:1-5,be44d87c-4f25-11e6-a0a8-525400de9ffd:1-15610498Sinkin0row)和store(store)数据转储回来后。sink和store（存储）sink操作可以支持过滤来自多个eventParser的数据。filter过滤器使用instance.properties中配置的filter，当然这个filter也可以在订阅的时候由渠道客户端设置。如果在客户端设置，则服务端配置文件instance.properties的配置将失效。sink之后，过滤后的数据会存储到eventStore中。目前eventStore的实现只有一个MemoryEventStoreWithBuffer，是一个基于内存的ringbuffer。使用这家商店有一个特点。这个ringbuffer是基于内存的，它的大小是有限制的（bufferSize=16*1024，也就是16M）。因此，当canal客户端消费缓慢时，ringbuffer满了就会阻塞sink操作，读取mysqlbinlog的eventParser线程也会阻塞。这种设计其实是合理的。因为canal的运行是pull模型，不是producerpush模型，不需要存储太多的数据，可以避免一些数据存储和持久化管理的问题。数据管理的复杂性大大降低。以上就是canal的parser线程的工作流程，主要对应从mysql中获取数据，做一些基本的收集过滤，然后存入内存。Binlog的consumercanal主要是想给mysql订阅binlog后的consumer使用。那么binlog是什么时候消费的呢？这是另一个线程。就像我们做一个toC系统，管理系统是必须的，用户使用的app或者web是另外一套。eventParser线程就像一个管理系统，基本的数据都被录入其中。canal的客户端就像是app端，是这些数据的消费者。binlog的主要消费者是canal的客户端。使用的协议是基于tcp的google.protobuf。当然tcp的模式是io多路复用，也就是nio。当我们的客户端发起请求时，服务器端会把eventStore中的数据传给客户端。根据客户端的ack机制，周期性的将binlog的元数据信息同步到zookeeper。Canal目录结构配置父目录：如下所示─canal.properties├──gamer---目录├──ww_social---目录├──wother---目录├──nihao---目录├──liveim---目录├──logback.xml├──spring---目录├──ym---目录└──xrm_ppp---目录这是完全展开的目录COPYcanal├──bin│├──canal.pid│├──启动。bat│├──startup.sh│└──stop.sh└──conf├──canal.properties├──game_center│└──instance.properties├──ww_social│├──h2.mv.db│├──h2.trace.db│└──instance.properties├──wwother│├──h2.mv.db│└──instance.properties├──nihao│├──h2.mv.db│├──h2.trace.db│└──instance.properties├──电影│├──h2.mv.db│└──instance.properties├──logback.xml├──spring│├──default-instance.xml│├──file-instance.xml│├──group-instance.xml│├──local-instance.xml│├──memory-instance.xml│└──tsdb│├──h2-tsdb.xml│├──mysql-tsdb.xml│├──sql│└──sql-map└──ym└───instance.propertiesCanal应用场景同步缓存redis/全文搜索EScanal一个常见的应用场景是同步缓存/全文搜索。当数据库发生变化时，通过binlog进行缓存/ES增量更新。当cache/ES更新出现问题时，应该回滚Binlog重新同步到过去的某个位置，并提供了全量刷新cache/ES的方法，如下图。发布任务 另一个常见的应用场景是发布任务。当数据发生变化时，需要通知其他依赖系统。原理是任务系统监听数据库变化，然后将变化的数据写入MQ/kafka进行任务下发。例如商品数据发生变化后，需要通知商品详情页、列表页、搜索页等先关闭系统。这种方式可以保证数据传递的准确性，这是通过MQ发送消息通知变化缓存所无法实现的，业务系统也不会散落着各种MQ代码，从而实现集合传递，如下图。数据异构在大型网站架构中，DB会使用分库分表来解决容量和性能问题，但分库分表后又会带来新的问题。比如不同维度的查询，或者聚合查询，这时候就会非常棘手。一般我们会通过数据异构机制来解决这个问题。所谓数据异构，是指需要连接查询的多张表按照一定的维度聚合在一个DB中。让你查询。Canal是实现数据异构的手段之一。本文由传智教育博学谷狂野建筑师教研团队发布。如果本文对您有帮助，请关注并点赞；有什么建议也可以留言或私信。您的支持是我坚持创作的动力。转载请注明出处！
