MySQL主从同步，及Canal实时数据库同步

╰+哭是因爲堅強的太久メ 2022-12-16 09:25 376阅读 0赞

# （一）MySQL主从同步 #

# **1.简介** #

Linux下MySQL数据库的主从同步用来实现读写分离。主数据库进行数据的插入，删除与更新；从数据库专门用来查询操作，缓解数据库的压力。

# 2.主从同步原理 #

Mysql主从同步是基于二进制日志机制，master通过二进制日志来记录数据库的变动情况，slave通过读取和执行该日志文件来保持和主服务器的数据一致。

Mysql主从同步，通常采用异步复制，从master复制到slave，由三个线程来完成。其中sql线程和IO线程在slave端，另一个IO线程在master端，要实现MySQL的replication首先需要打开master端的二进制log功能:

(1) master 将操作记录到二进制日志(binary log)中；  
(2) master有一个I/O线程将二进制日志发送到slave；  
(3) slave IO 线程 将master的binary log events读写到它的转送日志(relay log)；  
(4) slave SQL进程读取relay log，将变更记录数据到slave数据库中。

### 结构图 ###

![3f680eb7f2a3b7c1878241a54b3ad493.png][]

## ##

### 3. 配置主从同步的基本步骤 ###

1.  在主服务器上，必须开启二进制日志机制和配置一个独立的ID
2.  在每一个从服务器上，配置一个唯一的ID，创建一个用来专门复制主服务器数据的账号
3.  在开始复制进程前，在主服务器上记录二进制文件的位置信息
4.  如果在开始复制之前，数据库中已经有数据，就必须先创建一个数据快照（可以使用mysqldump导出数据库，或者直接复制数据文件）
5.  配置从服务器要连接的主服务器的IP地址和登陆授权，二进制日志文件名和位置

# **4.主从数据库配置** #

## ** 1.操作步骤** ##

1）开启master的二进制日志

2）开启slave的二进制日志

3）将slave指向master

4）开始复制

## ** 2.开启master二进制日志** ##

1）编辑mysql配置文件

1 [root@ ~]# vi /etc/my.cnf

2）添加二进制日志配置，开启二进制（mysql-bin只是二进制日志名称，可以自行指定）

1 server-id=1             #id是一定要指定的，是唯一的标识（master数据库要比slave数据库的id优先级高才行）
    2 log-bin=mysql-bin       #开启二进制日志

3）授权

登录数据库

需要给slave数据库配置一个用户/密码的权限

1 mysql> grant replication slave on *.* to 'root'@'slave数据库ip' identified by '密码';

允许某个ip地址的某个用户以某个密码对当前数据库的所有库和表进行复制操作

配置之后需要刷新权限

1 mysql> flush privileges;

上面修改配置文件需重启服务

1 [root@ ~]# service mysqld restart

4）查看master的状态

登录数据库

1 mysql> show master status;

![f31a624907b8d812a7dc066ffca30314.png][]

file：是日志文件名称

position：日志所在位置

## ** 3.开启slave的二进制日志** ##

登录slave服务器

1）配置my.cnf配置文件

1 [root@ ~]# vi /etc/my.cnf

2）添加slave二进制日志配置，开启二进制（mysql-bin只是二进制日志名称，可以自行指定）

1 server-id=2
    2 log-bin=mysql-bin

注意：每一台指定唯一的一个server-id标识

修改完配置服务需重启服务

1 [root@VM_0_16_centos ~]# service mysqld restart

3）配置slave指向master

1 mysql> change master to
    2     -> master_host='master ip',
    3     -> master_user='master 账号',
    4     -> master_password='授权密码',
    5     -> master_log_file='master日志文件',
    6     -> master_log_pos=master日志所在位置;

注意：`master_log_file、master_log_pos俩个参数,可以在master上执行show master status可以获取;`

## **4.开启主从复制** ##

在slave服务器上执行

1 mysql> start slave;

查看slave运行状态

mysql> show slave status\G;
    *************************** 1. row ***************************
                   Slave_IO_State: Waiting for master to send event
                      Master_Host: 106.53.73.200
                      Master_User: root
                      Master_Port: 3306
                    Connect_Retry: 60
                  Master_Log_File: mysql-bin.000001
              Read_Master_Log_Pos: 154
                   Relay_Log_File: VM_0_16_centos-relay-bin.000002
                    Relay_Log_Pos: 320
            Relay_Master_Log_File: mysql-bin.000001
                 Slave_IO_Running: Yes
                Slave_SQL_Running: Yes
                  Replicate_Do_DB:
              Replicate_Ignore_DB:
               Replicate_Do_Table:
           Replicate_Ignore_Table:
          Replicate_Wild_Do_Table:
      Replicate_Wild_Ignore_Table:
                       Last_Errno: 0
                       Last_Error:
                     Skip_Counter: 0
              Exec_Master_Log_Pos: 154
                  Relay_Log_Space: 536
                  Until_Condition: None
                   Until_Log_File:
                    Until_Log_Pos: 0
               Master_SSL_Allowed: No
               Master_SSL_CA_File:
               Master_SSL_CA_Path:
                  Master_SSL_Cert:
                Master_SSL_Cipher:
                   Master_SSL_Key:
            Seconds_Behind_Master: 0
    Master_SSL_Verify_Server_Cert: No
                    Last_IO_Errno: 0
                    Last_IO_Error:
                   Last_SQL_Errno: 0
                   Last_SQL_Error:
      Replicate_Ignore_Server_Ids:
                 Master_Server_Id: 1
                      Master_UUID: f8100725-c3e5-11e9-ae45-525400da2f1f
                 Master_Info_File: /var/lib/mysql/master.info
                        SQL_Delay: 0
              SQL_Remaining_Delay: NULL
          Slave_SQL_Running_State: Slave has read all relay log; waiting for more updates
               Master_Retry_Count: 86400
                      Master_Bind:
          Last_IO_Error_Timestamp:
         Last_SQL_Error_Timestamp:
                   Master_SSL_Crl:
               Master_SSL_Crlpath:
               Retrieved_Gtid_Set:
                Executed_Gtid_Set:
                    Auto_Position: 0
             Replicate_Rewrite_DB:
                     Channel_Name:
               Master_TLS_Version:
    1 row in set (0.00 sec)
    
    ERROR:
    No query specified

若看见 IO、SQL俩个线程状态为yes，则开启成功

1 Slave_IO_Running: Yes     #表示slave的日志读取线程开启
    2 Slave_SQL_Running: Yes    #表示SQL执行线程开启

# （二）Canal实时数据库同步 #

## 1.1 canal介绍 ##

Canal是一个基于MySQL二进制日志的高性能数据同步系统。Canal广泛用于阿里巴巴集团（包括[https://www.taobao.com][https_www.taobao.com]），以提供可靠的低延迟增量数据管道，github地址：[https://github.com/alibaba/canal][https_github.com_alibaba_canal]

Canal Server能够解析MySQL binlog并订阅数据更改，而Canal Client可以实现将更改广播到任何地方，例如数据库和Apache Kafka。

它具有以下功能：

*  支持所有平台。
 *  支持由Prometheus提供支持的细粒度系统监控。
 *  支持通过不同方式解析和订阅MySQL binlog，例如通过GTID。
 *  支持高性能，实时数据同步。（详见Performance）
 *  Canal Server和Canal Client都支持HA / Scalability，由Apache ZooKeeper提供支持
 *  Docker支持。

缺点：

不支持全量更新，只支持增量更新。

完整wiki地址：[https://github.com/alibaba/canal/wiki][https_github.com_alibaba_canal_wiki]

## 1.2 运作原理 ##

原理很简单：

1.  Canal模拟MySQL的slave的交互协议，伪装成mysql slave，并将转发协议发送到MySQL Master服务器。
2.  MySQL Master接收到转储请求并开始将二进制日志推送到slave（即canal）。
3.  Canal将二进制日志对象解析为自己的数据类型（原始字节流）

![1204b1d98674579cee101eb21845a4ad.png][]

# #

## 安装配置canal ##

在这里https://github.com/alibaba/canal/releases，下载canal releases包，如https://github.com/alibaba/canal/releases/download/canal-1.0.22/canal.deployer-1.0.22.tar.gz。

打开conf/example下的instance.properties文件，编辑

#################################################
    ## mysql serverId
    canal.instance.mysql.slaveId = 1234
    
    # position info
    canal.instance.master.address = 127.0.0.1:3306
    canal.instance.master.journal.name =  
    canal.instance.master.position = 
    canal.instance.master.timestamp = 
    
    #canal.instance.standby.address = 
    #canal.instance.standby.journal.name =
    #canal.instance.standby.position = 
    #canal.instance.standby.timestamp = 
    
    # username/password
    canal.instance.dbUsername = canal
    canal.instance.dbPassword = canal
    canal.instance.defaultDatabaseName = test
    canal.instance.connectionCharset = UTF-8
    
    # table regex
    canal.instance.filter.regex = .*\\..*
    # table black regex
    canal.instance.filter.black.regex =  
    
    #################################################

## (2)编辑Mysql配置文件 ##

vi /etc/my.cnf
    # 加入以下内容
    server_id=1
    log_bin = mysql-bin
    binlog_format = ROW
    expire_logs_days = 30
    #重启mysql服务
    systemctl restart mysqld #或重启docker容器

再次使用命令show variables like 'log\_%';进行查看，为ON表明binlog开启

![3fc9ea4e574a75b9ae344afd2fb3095f.png][]

## (3) 添加同步数据专用账号 ##

#创建账号
    CREATE USER canal IDENTIFIED WITH MYSQL_NATIVE_PASSWORD BY 'canal';
    #给账号赋权限
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; 
    #刷新
    FLUSH PRIVILEGES;

然后启动canal/bin/startup.sh

## （4）canal客户端编写 ##

服务端启动完毕后，在客户端即可监听test库的变化。

新建一个java maven项目，pom.xml里添加依赖

<dependency>
          <groupId>com.alibaba.otter</groupId>
          <artifactId>canal.client</artifactId>
          <version>1.0.12</version>
        </dependency>

package com.data.gx.config;
    
    
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.protocol.CanalEntry;
    import com.alibaba.otter.canal.protocol.Message;
    import com.data.gx.constant.GXTable;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.context.event.ApplicationStartedEvent;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.annotation.PropertySource;
    import org.springframework.stereotype.Component;
    import java.net.InetSocketAddress;
    import java.util.List;
    
    @Component
    @PropertySource("classpath:canal.properties")
    public class EventConfig implements ApplicationListener<ApplicationStartedEvent> {
    
        @Value("${host}")
        private String host;
        @Value("${port}")
        private int port;
        @Value("${destination}")
        private String destination;
        @Value("${database}")
        private String database;
    
        /**
         * 连接canal，并监听数据库，获取日志数据
         * @param event
         */
        public void onApplicationEvent(ApplicationStartedEvent event) {
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), destination, "", "");
    
            while (true){
                try {
                    //批量接收数据量大小
                    int batchSize = 1000;
                    //Canal连接处理
                    connector.connect();
                    System.out.println("======== Connected canal server success ========");
                    //订阅数据库名
                    connector.subscribe(database+"\\..*");
                    connector.rollback();
                    while (true) {
                        Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                        long batchId = message.getId();
                        int size = message.getEntries().size();
                        if (batchId == -1 || size == 0) {
                            try {
                                //1分钟秒获取一次数据
                                Thread.sleep(60000);
                            } catch (InterruptedException e) {
                            }
                        } else {
                            dealEntry(message.getEntries());
                        }
                        connector.ack(batchId); // 提交确认
                    }
                }catch (Exception e){
    //                e.printStackTrace();
                    System.out.println("!!!!!! Disconnect canal server  ！！！！！！");
                }finally {
                    connector.disconnect();
                    try {
                        //连接失败后，15秒重连一次
                        Thread.sleep(15000);
                        System.out.println("...... Connecting canal server  ......");
                    } catch (InterruptedException e) {}
    
                }
            }
    
    
        }
    
    
        //处理日志数据
        private  void dealEntry(List<CanalEntry.Entry> entrys) {
            for (CanalEntry.Entry entry : entrys) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    continue;
                }
    
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
                //获取事件类型
                CanalEntry.EventType eventType = rowChage.getEventType();
    //            System.out.println(String.format("======binlog[%s:%s] , database,table : %s,%s] , eventType : %s",
    //                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
    //                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
    //                    eventType));
                //获取触发事件的表名
                String tableName =  entry.getHeader().getTableName();
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                    //校验事件类型
                    if (eventType == CanalEntry.EventType.DELETE) {
    //                    printColumn(rowData.getBeforeColumnsList());
    
                    } else if (eventType == CanalEntry.EventType.INSERT) {
    //                    printColumn(rowData.getAfterColumnsList());
                    } else if (eventType == CanalEntry.EventType.UPDATE){
    //                    System.out.println("-------> before");
    //                    printColumn(rowData.getBeforeColumnsList());
    //                    System.out.println("-------> after");
    //                    printColumn(rowData.getAfterColumnsList());
    
                    }
                }
            }
        }
    
    }

注意newSingleConnector那里，设置canal server的ip和端口，端口默认为11111。example是和conf目录下的相对应。

[3f680eb7f2a3b7c1878241a54b3ad493.png]: /images/20221123/0594c1ee33714e53abe6f1c1eea2810e.png
[f31a624907b8d812a7dc066ffca30314.png]: /images/20221123/d25b1ea1cf8d45429128dd97706f8936.png
[https_www.taobao.com]: https://links.jianshu.com/go?to=https%3A%2F%2Fwww.taobao.com%2F
[https_github.com_alibaba_canal]: https://links.jianshu.com/go?to=https%3A%2F%2Fgithub.com%2Falibaba%2Fcanal
[https_github.com_alibaba_canal_wiki]: https://links.jianshu.com/go?to=https%3A%2F%2Fgithub.com%2Falibaba%2Fcanal%2Fwiki
[1204b1d98674579cee101eb21845a4ad.png]: /images/20221123/57b0dc8b0e5842afa6d34446b6a5fff0.png
[3fc9ea4e574a75b9ae344afd2fb3095f.png]: /images/20221123/52b8e59620de462e9c12d70261a9f65a.png