MySQL主从复制
mysql服务端修改配置并重启

问题：创建canal用户的目的是什么？直接使用现有的用户名可以吗，比如root。

答案：有些用户没有REPLICATION SLAVE, REPLICATION CLIENT的权限，用这些用户连接canal时，无法获取到binlog。

这里的canal用户授权了全部权限，所以客户端可以从canal中获取binlog。
明确两个概念：canal server连接mysql，客户端连接canal server。

canal指的是canal server，它会读取mysql的binlog，解析后存储起来
客户端指的是消费canal server的binlog
本机连接服务端，验证binlog的格式是ROW

mysql主从复制的原理：

master将改变记录到二进制日志(binary log)中；
slave将master的binary log events拷贝到它的中继日志(relay log)；
slave重做中继日志中的事件，将改变反映它自己的数据。

在启动canal之前，先来了解下什么是mysql的binlog:

mysql数据文件下会生成mysql-bin.xxx的binlog文件，以及索引文件

总用量 26228

srwxrwxrwx 1 mysql mysql 0 10月 11 09:57 mysql.sock
针对mysql的操作都会有二进制的事件记录到binlog文件中。下面的一些操作包括创建用户，授权，创建数据库，创建表，插入一条记录。

create database

===》创建数据库

create table test ( uid int (4) primary key not null auto_increment, name varchar(10) not null) ==》创建表
canal_test
BEGIN ==》插入记录，这里有事务。但是没有把具体的语句打印出来

部署canal server到6.52，并启动。查看canal的日志：

查看instance的日志：

canal server的conf下有几个配置文件

先来看canal.properties的common属性前四个配置项：

canal.id是canal的编号，在集群环境下，不同canal的id不同，注意它和mysql的server_id不同。
ip这里不指定，默认为本机，比如上面是192.168.6.52，端口号是11111。zk用于canal cluster。

再看下canal.properties下destinations相关的配置：

这里的canal.destinations = example可以设置多个，比如example1,example2，
则需要创建对应的两个文件夹，并且每个文件夹下都有一个instance.properties文件。

全局的canal实例管理用spring，这里的file-instance.xml最终会实例化所有的destinations instances:

比如canal.instance.destination等于example，就会加载example/instance.properties配置文件

example下instance.properties配置文件不需要修改。一个canal server可以运行多个canal instance。

在mysql上创建数据库，创建表，插入一条记录，再修改记录。

修改客户端测试例子的连接信息。其中example对应了canal实例的名称。

注意：如果连接有错误，客户端测试例子会立即结束，打印## stop the canal client。正常的话，终端不会退出，会一直运行。

SimpleCanalClientTest控制台的结果如下：

插入一条记录：（其中uid和name的update都等于true）

修改记录：（其中name的update等于true）

canal安装包下的example instance下除了example.log外，还有一个meta.log

canal client与canal server之间是C/S模式的通信，客户端采用NIO，服务端采用Netty。
canal server启动后，如果没有canal client，那么canal server不会去mysql拉取binlog。

canal client与canal server之间属于增量订阅/消费，流程图如下：（其中C端是canal client，S端是canal server）

canal client调用connect()方法时，类型为PacketType.HANDSHAKE，接着写入CLIENTAUTHENTICATION。然后调用subscribe()方法，类型为SUBSCRIPTION。

对应服务端采用netty处理RPC请求（CanalServerWithNetty）:

ClientAuthenticationHandler处理鉴权后，会移除HandshakeInitializationHandler和ClientAuthenticationHandler

以client发送GET，server从mysql得到binlog后，返回MESSAGES给client为例，说明client和server的rpc交互过程：

SimpleCanalConnector发送GET请求，并读取响应结果的流程：

服务端SessionHandler处理客户端发送的GET请求流程：

get/ack/rollback协议介绍：

Message getWithoutAck(int batchSize)，允许指定batchSize，一次可以获取多条，每次返回的对象为Message，包含的内容为：
– batch id 唯一标识
– entries 具体的数据对象，对应的数据对象格式：EntryProtocol.proto
void rollback(long batchId)，顾命思议，回滚上次的get请求，重新获取数据。基于get获取的batchId进行提交，避免误操作
void ack(long batchId)，顾命思议，确认已经消费成功，通知server删除数据。基于get获取的batchId进行提交，避免误操作
EntryProtocol.protod对应的canal消息结构如下：

logfileName [binlog文件名]

executeTime [binlog里记录变更发生的时间戳,精确到秒]

eventType [insert/update/delete类型]
entryType [事务头BEGIN/事务尾END/数据ROWDATA]
storeValue [byte数据,可展开，对应的类型为RowChange]

isDdl [是否是ddl变更操作，比如create table/drop table]
sql [具体的ddl sql]
rowDatas [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]
beforeColumns [Column类型的数组，变更前的数据字段]
afterColumns [Column类型的数组，变更后的数据字段]

isKey [是否为主键]
updated [是否发生过变更]
isNull [值是否为null]
value [具体的内容，注意为string文本]
SessionHandler中服务端处理客户端的其他类型请求，都会调用CanalServerWithEmbedded的相关方法：

CanalServer包含多个Instance，它的成员变量canalInstances记录了instance名称与实例的映射关系。
因为是一个Map，所以同一个Server不允许出现相同instance名称，比如不能同时有两个example在一个server上。

下图表示一个server有两个instance，每个Client连接一个Instance。
每个Canal实例模拟为一个MySQL的slave，所以每个Instance的slaveId必须不一样。比如图中两个Instance的id分别是1234和1235。

注意这里每个Canal Client都对应一个Instance，每个Client在启动时，都会指定一个Destination，这个Destination就表示Instance的名称。
所以CanalServerWithEmbedded处理各种请求时的参数都有ClientIdentity，从ClientIdentity中获取destination，就可以获取出对应的CanalInstance

下面以CanalServerWithEmbedded的订阅方法为例：

每个CanalInstance中包括了四个组件：EventParser、EventSink、EventStore、MetaManager。

服务端主要的处理方法包括get/ack/rollback，这三个方法都会用到Instance上面的几个内部组件，主要还是EventStore和MetaManager：

在这之前，要先理解EventStore的含义，EventStore是一个RingBuffer，有三个指针：Put、Get、Ack。

Put: Canal Server从MySQL拉取到数据后，放到内存中，Put增加
Get: 消费者（Canal Client）从内存中消费数据，Get增加
Ack: 消费者消费完成，Ack增加。并且会删除Put中已经被Ack的数据
这三个操作与Instance组件的关系如下：

客户端通过canal server获取mysql binlog有几种方式（get方法和getWithoutAck）：

如果timeout为null，则采用tryGet方式，即时获取
如果timeout不为null
timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少

注意：EventStore的实现采用了类似Disruptor的RingBuffer环形缓冲区。RingBuffer的实现类是MemoryEventStoreWithBuffer

get方法和getWithoutAck方法的区别是：

get方法会立即调用ack
getWithoutAck方法不会调用ack

以10条数据为例，初始时current=-1，第一个元素起始next=0，end=9，循环[0,9]所有元素。
List元素为(A,B,C,D,E,F,G,H,I,J)

第一批10个元素put完成后，putSequence设置为end=9。假设第二批又Put了5个元素:(K,L,M,N,O)

current=9，起始next=9+1=10，end=9+5=14，在Put完成后，putSequence设置为end=14。

这里假设环形缓冲区的最大大小为15个（源码中是16MB），那么上面两批一共产生了15个元素，刚好填满了环形缓冲区。
如果又有Put事件进来，由于环形缓冲区已经满了，没有可用的slot，则Put操作会被阻塞，直到被消费掉。

下面是Put填充环形缓冲区的代码，检查可用slot（checkFreeSlotAt方法）在几个put方法中。

Put是生产数据，Get是消费数据，Get一定不会超过Put。比如Put了10条数据，Get最多只能获取到10条数据。但有时候为了保证Get处理的速度，Put和Get并不会相等。
可以把Put看做是生产者，Get看做是消费者。生产者速度可以很快，消费者则可以慢慢地消费。比如Put了1000条，而Get我们只需要每次处理10条数据。

仍然以前面的示例来说明Get的流程，初始时current=-1，假设Put了两批数据一共15条，maxAbleSequence=14，而Get的BatchSize假设为10。
初始时next=current=-1，end=-1。通过startPosition，会设置next=0。最后end又被赋值为9，即循环缓冲区[0,9]一共10个元素。

ack操作的上限是Get，假设Put了15条数据，Get了10条数据，最多也只能Ack10条数据。Ack的目的是清空缓冲区中已经被Get过的数据

rollback回滚方法的实现则比较简单，将getSequence回退到ack位置。

下图展示了RingBuffer的几个操作示例：

EventStore负责存储解析后的Binlog事件，而解析动作负责拉取Binlog，它的流程比较复杂。需要和MetaManager进行交互。
比如要记录每次拉取的Position，这样下一次就可以从上一次的最后一个位置继续拉取。所以MetaManager应该是有状态的。

EventParser的流程如下：

Connection获取上一次解析成功的位置 (如果第一次启动，则获取初始指定的位置或者是当前数据库的binlog位点)
Connection建立链接，发送BINLOG_DUMP指令
Mysql开始推送Binaly Log
接收到的Binaly Log的通过Binlog parser进行协议解析，补充一些特定信息
传递给EventSink模块进行数据存储，是一个阻塞操作，直到存储成功
存储成功后，定时记录Binaly Log位置

上面提到的Connection指的是实现了ErosaConnection接口的MysqlConnection。
EventParser的实现类是实现了AbstractEventParser的MysqlEventParser。

EventParser解析binlog后通过EventSink写入到EventStore，这条链路可以通过EventStore的put方法串联起来：

其实这里还有一个EventTransactionBuffer缓冲区，即Parser解析后先放到缓冲区中，
当事务发生时或者数据超过阈值，就会执行刷新操作：即消费缓冲区的数据，放到EventStore中。
这个缓冲区有两个偏移量指针：putSequence和flushSequence。

单机模拟两个Canal Server，将单机模式复制出两个文件夹，并修改相关配置

启动canal_m

启动canal_s

master提供服务，canal_m/logs/example/example.log下有日志，而canal_s/logs没有example文件夹

查看Canal HA记录在ZK的信息

启动example的ClusterCanalClientTest

执行SQL：update test set name = 'zqh' where uid=1;，控制台打印日志如下：

再次查看ZK中记录的客户端信息：

一个Instance对应一个Client，这里的Instance名称为example，对应的客户端编号是1001
为了验证Instance确实是由指定的Client连接，在Server上查看11112端口

"postion":{"included":false,"journalName":"mysql-bin.000004","position":1331,"serverId":1,"timestamp":1507791918000}} ==》serverId表示MySQL的server_id

tcp 0 0 0.0.0.0:11112 0.0.0.0:* LISTEN 27816/java ==》Canal服务端
tcp 0 19 192.168.6.52:11112 10.57.241.44:53942 ESTABLISHED 27816/java ==》Canal客户端
停止canal_m

Instance会在slave节点即canal_s上启动

停止canal_m后，只剩下canal_s，所以Canal集群只有一个节点了：

切换过程中，Client的日志

再次执行SQL语句

停止客户端后，查询ZK中的客户端信息。注意，仍然有cursor信息，但是没有running，因为instance没有对应的client了。

cursor信息是instance消费binlog的位置，即使客户端停掉了，也仍然保留在zk中。

注意：1001是ClientIdentity的固定编号，相关源码在SimpleCanalConnector的构造方法里。

下面总结下zk中的相关记录：

下图是Canal Server HA的流程图：

canal server要启动某个canal instance时都先向zookeeper进行一次尝试启动判断 (实现：创建EPHEMERAL节点，谁创建成功就允许谁启动)
创建zookeeper节点成功后，对应的canal server就启动对应的canal instance，没有创建成功的canal instance就会处于standby状态
一旦zookeeper发现canal server A创建的节点消失后，立即通知其他的canal server再次进行步骤1的操作，重新选出一个canal server启动instance.
canal client每次进行connect时，会首先向zookeeper询问当前是谁启动了canal instance，然后和其建立链接，一旦链接不可用，会重新尝试connect.

Canal Client的方式和canal server方式类似，也是利用zookeeper的抢占EPHEMERAL节点的方式进行控制。

关于Canal Client HA的验证，可以参考：blog.csdn.net/xiaolinzi00…

在IDEA中同时启动多个客户端，执行一条SQL语句，其中一个客户端会打印日志，另一个不会打印。
停止该客户端。
再次执行SQL语句，另外一个客户端会打印日志
Client1的日志：

停止Client1后，Client2的日志：

观察ZK节点中instance对应的client节点，在Client切换时，会进行变更。
比如下面的客户端从56806端口切换到了56842端口。
把所有客户端都关闭后，1001下没有running。表示instance没有客户端消费binlog了。

具体实现相关类有：ClientRunningMonitor/ClientRunningListener/ClientRunningData。

client running相关控制，主要为解决client自身的failover机制。
canal client允许同时启动多个canal client，通过running机制，可保证只有一个client在工作，其他client做为冷备.
当运行中的client挂了，running会控制让冷备中的client转为工作模式，
这样就可以确保canal client也不会是单点. 保证整个系统的高可用性.

下图左边是客户端的HA实现，右边是服务端的HA实现

先理解：github.com/alibaba/can…

重新看下CanalServerWithEmbedded的订阅方法。我们知道客户端在连接服务端的某个destination之后，会紧接着调用subscribe()方法。

客户端连接服务端时，必须指定destination名称，因为一个服务端可能有多个destination。比如服务端启动了两个Instance，它们的destination名称分别是example1和example2。
假设有两个客户端A和B，A连接example1，B连接example2。 服务端的canalInstances字典为：{example1=>Instance1，example2->Instance2}。
那么ClientA的destination等于example1，对应的服务端实例为Instance1。ClientB的destination等于example2，对应的服务端实例为Instance3。

* 客户端订阅，重复订阅时会更新对应的filter信息

这里面关于订阅方法有两个地方，CanalInstance本身调用了subscribeChange，它关联的MetaManager也调用了subscribe方法。

一个CanalServer可以有多个CanalInstance，每个Instance都会有一个MetaManager。而一个Instance对应一个Client。
那么，这么说来，一个MetaManager也就只会有一个Client了。但是从下面的数据结构来看的话，一个MetaManager貌似可以有多个Destination。

猜测：多个Client可以连接到同一个Instance（虽然只会有一个Instance起作用），所以一个MetaManager可以管理多个Client。

NO！Client的HA与MetaManager记录的Client是不一样的。HA表示同一时间只有一个Client起作用，那么MetaManager不可能同时记录两个Client。
官方ClientAPI文档上：ClientIdentity是canal client和server交互之间的身份标识，目前clientId写死为1001.
(目前canal server上的一个instance只能有一个client消费，clientId的设计是为1个instance多client消费模式而预留的，暂时不需要理会)

也就是说：一个Instance还是有可能有多个Client连接上来的。

这里的数据结构为什么这么设计，还需要参考AbstractMetaManagerTest的doSubscribeTest方法来理解。

对于相同的destination，可以订阅不同的client。下面的示例分别订阅了[client1,client2]和[client1,client3]。

CanalServerWithEmbedded的订阅方法最后还会调用AbstractCanalInstance的subscribeChange方法。
这里会设置表名的filter，以及黑名单。配置项在instance.properties中。

filter表示客户端要通过Canal Server获取MySQL哪些表的binlog，上面配置项表示获取所有表。

对应在EventParser中，存在两个Filter的引用。比如上面eventParser.setEventFilter()方法会设置AbstractEventParser的eventFilter。

AbstractEventParser的start()方法是解析binlog的主要方法。在启动transactionBuffer和BinLogParser后，会启动一个后台的工作线程parseThread一直运行：

注意：下面的几个步骤是嵌套在一个while死循环里，最后会进行sleep。

这里的erosaConnection指的是Canal Server到MySQL的连接。
而前面我们说的客户端（CanalClient）连接CanalConnector指的是CanalClient到CanalServer的连接。

CanalServer到MySQL的连接是要获取binlog的dump数据包。而CanalClient到CanalServer有多种请求（GET/ACK等）。

我们不会具体分析dump的流程，不过看下erosaConnection的MySQL实现MysqlConnection是如何在获取到事件后调用回调函数。

服务端有一个心跳线程，它的目的是消费transactionBuffer，并写入到EventSink中。

EventSink最终会将数据写入到EventStore中，即Put到RingBuffer中。