如何基于canal實現mysql的數據同步

本篇內容介紹了“如何基于canal實現MySQL的數據同步”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

創新互聯主營東川網站建設的網絡公司,主營網站建設方案,app軟件開發公司,東川h5微信平臺小程序開發搭建,東川網站營銷推廣歡迎東川等地區企業咨詢

canal是什么？

canal [k?'n?l]，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費

基于日志增量訂閱和消費的業務包括

數據庫鏡像
數據庫實時備份
索引構建和實時維護(拆分異構索引、倒排索引等)
業務 cache 刷新
帶業務邏輯的增量數據處理

當前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

工作原理

如何基于canal實現mysql的數據同步

基于上面的講解，我們在實現canal之前，先簡單做一個主從復制。一主一從

首先下載mysql 鏡像,并啟動

docker pull mysql:latest
docker run -itd --name mysql-1 -p 23306:3306 -e MYSQL_ROOT_PASSWORD=root  mysql
docker run -itd --name mysql-2 -p 23307:3306 -e MYSQL_ROOT_PASSWORD=root  mysql

相關命令再解釋一下： name xxx :xxx為容器名 p 111:222 其中111是宿主機端口，222是容器端口 MYSQL_ROOT_PASSWORD=root 設置root賬戶密碼為root
進入容器測試一下，一切正常
設置 mysql-1為主，mysql-2為從庫
修改一下 mysql的配置，安裝vim編輯器

apt-get update
apt-get install vim

在主庫創建一個mysql賬戶給從庫使用

CREATE USER 'slave'@'%' IDENTIFIED BY '123456';
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON . TO 'slave'@'%';
FLUSH PRIVILEGES；

如何基于canal實現mysql的數據同步

修改一下從服務器

[mysqld]
pid-file        = /var/run/mysqld/mysqld.pid
socket          = /var/run/mysqld/mysqld.sock
datadir         = /var/lib/mysql
secure-file-priv= NULL
server_id=100
log-bin=mysql-slave-bin
relay_log=edu-mysql-relay-bin

退出重啟從服務器docker
進入從服務器執行

mysql> change master to master_host='172.17.0.4', master_user='slave', master_password='123456', master_port=3306, master_log_file='edu-mysql-bin.000001', master_log_pos= 877, master_connect_retry=30;

相關命令解釋 master_port：Master的端口號，指的是容器的端口號 master_user：用于數據同步的用戶 master_password：用于同步的用戶的密碼 master_log_file：指定 Slave 從哪個日志文件開始復制數據，即上文中提到的 File 字段的值 master_log_pos：從哪個 Position 開始讀，即上文中提到的 Position 字段的值 master_connect_retry：如果連接失敗，重試的時間間隔，單位是秒，默認是60秒在Slave 中的mysql終端執行show slave status \G;用于查看主從同步狀態。

出現一下信息說明配置成功
接下來再主庫寫數據，從庫同步成功
簡單的主從同步完成了，但是我們要想，怎么實現的主從同步，對吧；其實就是通過同步二進制日志文件，從服務器會起一個io進程，讀取二進制文件同步到從服務器
簡單看一下二進制文件的內容；

如何基于canal實現mysql的數據同步

為什么再將canal 之前要先說主從復制呢，其實canal 就是把自己偽裝成了從服務器，從而讀取日志，拿到數據；

使用docker 部署canal

參考鏈接

docker pull canal/canal-server:latest
# 下載腳本
wget https://raw.githubusercontent.com/alibaba/canal/master/docker/run.sh 

# 構建一個destination name為test的隊列， address 對應的數據庫ip+端口 ，dbUsername對應數據庫用戶名，dbPassword對應數據庫密碼，注意修改為自己的
sh run.sh -e canal.auto.scan=false \
-e canal.destinations=test \
-e canal.instance.master.address=172.17.0.4:3306  \
-e canal.instance.dbUsername=canal  \
-e canal.instance.dbPassword=canal  \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false  \

啟動之后可以進入容器，看一下里面的日志,如果出現了標紅的信息，說明成功，否則就查看里面的報錯信息吧！也不難

如何基于canal實現mysql的數據同步

配合php 查看數據變化（此處不限php，java,go，python等都有接口）
多語言連接https://github.com/alibaba/canal

php 監聽數據變化

canal-php canal-php 是阿里巴巴開源項目 Canal是阿里巴巴mysql數據庫binlog的增量訂閱&消費組件的 php 客戶端。為 php 開發者提供一個更友好的使用 Canal 的方式。Canal 是mysql數據庫binlog的增量訂閱&消費組件。

基于日志增量訂閱&消費支持的業務：

數據庫鏡像
數據庫實時備份
多級索引 (賣家和買家各自分庫索引)
search build
業務cache刷新
價格變化等重要業務消息

關于 Canal 的更多信息請訪問 https://github.com/alibaba/canal/wiki

應用場景 canal-php 作為Canal的客戶端，其應用場景就是Canal的應用場景。關于應用場景在Canal介紹一節已有概述。舉一些實際的使用例子：

1.代替使用輪詢數據庫方式來監控數據庫變更，有效改善輪詢耗費數據庫資源。

2.根據數據庫的變更實時更新搜索引擎，比如電商場景下商品信息發生變更，實時同步到商品搜索引擎 Elasticsearch、solr等

3.根據數據庫的變更實時更新緩存，比如電商場景下商品價格、庫存發生變更實時同步到redis

4.數據庫異地備份、數據同步

5.根據數據庫變更觸發某種業務，比如電商場景下，創建訂單超過xx時間未支付被自動取消，我們獲取到這條訂單數據的狀態變更即可向用戶推送消息。

6.將數據庫變更整理成自己的數據格式發送到kafka等消息隊列，供消息隊列的消費者進行消費。

工作原理 canal-php 是 Canal 的 php 客戶端，它與 Canal 是采用的Socket來進行通信的，傳輸協議是TCP，交互協議采用的是 Google Protocol Buffer 3.0。
工作流程
使用組件安裝，此處我有一個laravel框架，直接在laravel里面安裝使用了，相關代碼貼出

 # 安裝組件canal-php
 composer require xingwenge/canal_php
# 編寫腳本監聽
<?php

namespace App\Console\Commands;
use xingwenge\canal_php\CanalClient;
use xingwenge\canal_php\CanalConnectorFactory;
use xingwenge\canal_php\Fmt;
use Illuminate\Console\Command;

ini_set('display_errors', 'On');
error_reporting(E_ALL);
class CanalDemo extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'CanalDemo';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '測試canal';

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
    }

    /**
     * Execute the console command.
     *
     * @return mixed
     */
    public function handle()
    {
        try {
            $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);
            # $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SWOOLE);

            $client->connect("172.17.0.5", 11111);//此處修改為自己的配置
            $client->checkValid();
            $client->subscribe("1001", "test", ".*\\..*");//對應啟動容器時test的隊列名
            while (true) {
                $message = $client->get(100);
                if ($entries = $message->getEntries()) {
                    foreach ($entries as $entry) {
                        Fmt::println($entry);
                    }
                }
                sleep(1);
            }

            $client->disConnect();
        } catch (\Exception $e) {
            echo $e->getMessage(), PHP_EOL;
        }

    }

}