canal源码分析

Posted 2021-02-03 dhh-blog

本人阅读canal源码心得

      canal用来干嘛的?

      说的简单直白点就把你的数据库的binlog文件内容准实时传递给你的客户端，有了数据还不是想干嘛就干嘛。

      它的大致框架是什么呢?

      如果leader提出设计canal这样的需求，脑海中肯定浮现这样一个类CS的系统架构，其中S端就用来处理一些跟mysql，binlog以及其他一些杂七杂八的事，C端最多把接收的数据封装下，做下简单处理。至于怎么通信呢？那肯定首选Netty了。有个大致框架，接来下就是进一步细化。

     客户端、服务端的的信息管理。

     既然有客户端和服务端，那肯定要管理下这些元信息，信息管理无非就是把信息找个地方存储下，在目前框架中常见的就这几种：本地内存存储，本地文件存储，其他第三方存储容器。在canal里面采用本地内存+zookeeper存储方式。zookeeper中创建了如下几个节点（针对服务端）:

     1； /otter/canal/cluster.其中cluster的child节点分别存储的是各个S端的server信息,节点名称ip:port,属性:临时节点，

    2；/otter/canal/destinations/{destination}/cluster/{ip:port}   注册实例信息

    3；/otter/canal/destinations/{destiantion}/running/  存储的是正在运行的节点信息

    既然有了这些节点那肯定要在在节点上设置监听器，监听器主要是处理节点丢失和节点信息有变的情况。对于注册实例信息的节点，监听器主要是在session断连的情况下重新注册信息。对于running节点信息，监听器主要处理节点内容改变和节点删除，节点内容有变（例如主动下线某个实例）时，则Server端发起stop non-active 的实例操作，若节点删除了，则主动启动实例，并注册信息到running节点。

    如何取出binlog的记录并存储

    客户端和服务端的metadata信息处理完了，接下来处理的是如何把binlog的信息fetch出来，存储在本地。canal对此抽象出了三个组件:EventParser,EventSink,EventStore和MetaManager。其中MetaManager记录的是客户端的消费信息，主要为三种：destinations——Map数据结构，key为server端实例名称，value为订阅了该实例的消费客户端；cursors——Map数据结构，key为客户端的抽象类ClientIdentity,value为消费的位置信息；batches——记录消费的批次信息。MetaManager里面的信息会同步到Zookeeper中(针对ZookeeperMetaManager),其中zookeeper的对应的存储节点分别为:

     1,destiantions:/otter/canal/destinations/{destination}

     2,cursor:/otter/canal/destiantions/{destination}/cursor/{clientId}

     3,batches:/otter/canal/destiantions/{destination}/mark/{clientId}/{batchId},

一份信息两个地方存储，为了保证数据的一致性，启动了一个线程，定时更新本地内存的cursor信息到远程zookeeper中。

    EventSink类似于过滤器的作用，针对获取到的Bin log进行过滤。EventSink主要包含Handers（List数据结构），一些固定的过滤性配置:filterTransactionEntry(是否需要过滤事务头和尾部)，filterEmptyTransactionEntry(是否需要过滤空事务头和尾部)，一个过滤器filter。EventSink里面的组件功从组件的名字就能看出来了。

其中EventStore用于存储Bin log的记录，主要数据结构是数组和相关的位置index属性，通过一个数组和index属性（最新put的位置在哪，消费到哪了），构建了一个环形的队列。

其中EventParser主要两个功能，一个是按需fetch bin log记录，一个是解析bin log记录转化成CanalEntry。其中Bin log的位置获取，主要是通过查询 show master status 和show binlog events limit 1，支持查找最新的和查找某个时刻后的。当然在这个过程中还需要考虑种种异常情况，比如文件找不到怎么办，发生了准备切换怎么办（回退到fallbackIntervalInSeconds之前，按时间来fetch）。

   EventParser,EventStore,EventSink之间如何协调工作？

   如下面时序图所示（来源:https://www.jianshu.com/p/b50cbb7254b8）:

 

  如何接受客户端的请求

   S端的架构设计好了，接下来设计C端与S端的通信。Canal中的连接部分是基于Netty,通过在ChannelPipeline中添加FixedHeaderFrameDecoder,HandshakeInitializationHandler,ClientAuthenticationHandler,SessionHandler，从名称看也能猜出FixedHeaderFrameDecoder用于编码解码，HandshakeInitializationHandler用于处理请求连接的动作，ClientAuthenticationHandler那肯定是校验客户端了，核心的处理内容在SessionHandler。其中SessionHandler主要处理客户端的5中指令：SUBSCRIPTION,UNSUBSCRIPTION,GET,CLIENTACK,CLIENTROLLBACK。针对SUBSCRIPTION请求,Server记录客户端要订阅实例信息，以及客户端的过滤器信息。针对GET请求，Server从EventStore中获取Bin log记录信息，针对ACK请求， Server更新客户端的消费记录，针对CLIENTROLLBACK请求 ，客户端端消费失败了，Server将该客户端的消费进度回滚。

   客户端

  客户端设计时要解决得首要问题就是采用什么方式来消费Bin log记录。客户端采用的时候pull的方式，定时去拉取信息。

  系统框架图