Canal實現Mysql和ES數據同步

      1.ES概述

　　ES全稱為"ElasticSewrch", 是一個基于RESTful web接口, 并且構建在Apache Lucene之上的開源分布式搜索引擎。

　　ES可以在極短的時間內存儲、搜索和分析大量的數據。通常作為具有復雜搜索場景情況下的核心發(fā)動機。

　　簡單來說, ES可以幫助我們快速完成海量數據的搜索工作。

　　2.場景介紹

　　在實際開發(fā)中,為了保證數據的安全性和持久性, 所以,我們一般會把數據存儲的數據庫中, 比如mysql數據庫。

　　但, 如果數據量比較大, 并且搜索業(yè)務比較多, 則需要使用ES來實現我們的搜索功能。

　　而ES所實現的搜索, 需要把數據放入ES中, 才能實現, 所以, 就需要我們把mysql中的數據, 同步到ES中才可以。

　　3.實現方案

　　實現Mysql與ES之間的數據同步, 大體上可以分為三種方案:

　　方案一：同步調用

　　方案二：異步通知

　　方案三：監(jiān)聽binlog

　　4.監(jiān)聽binlog

　　在以上三種實現方案中, 可靠性相對來說比較好的是第三種方案"監(jiān)聽binlog",

　　| 方案 | 優(yōu)勢 | 缺點 |

　　| ---------- | ------------------------------ | -------------------------------------- |

　　| 同步調用 | 實現簡單，粗暴 | 業(yè)務耦合度高 |

　　| 異步通知 | 低耦合，實現難度一般 | 依賴mq的可靠性 |

　　| 監(jiān)聽binlog | 完全解除服務間耦合, 可靠性較強 | 開啟binlog增加數據庫負擔、實現復雜度高 |

　　4.1 實現原理

　　binlog其實就是Binary Log, 是MySQL二進制日志, 也可叫作變更日志(Update Log), 是 MySQL 中非常重要的日志。

　　主要用于記錄數據庫的變化情況，即 SQL語句的DDL和DML語句，不包含數據記錄查詢操作。

　　BinaryLog通常用于數據庫的主從復制. 其工作原理如下:

　　- 1)MySQL master 將數據變更寫入二進制日志( binary log)，其中記錄的數據叫做binary log events

　　- 2)MySQL slave 將 master 的 binary log events拷貝到它的中繼日志(relay log)

　　- 3)MySQL slave 重放 relay log 中事件，將數據變更反映它自己的數據

　　4.2 實現方案-Canal

　　Canal是阿里巴巴旗下的一款開源項目, 基于Java開發(fā)。Canal就是監(jiān)聽數據庫的binary log，從而提供增量數據訂閱&消費。

　　GitHub的地址：https://github.com/alibaba/canal

　　Canal就是把自己偽裝成MySQL的一個slave節(jié)點，從而監(jiān)聽master的binary log變化。

　　Canal會把得到的變化信息通知給Canal的客戶端，進而完成對其它數據庫的同步。

　　5.搭建環(huán)境

　　5.1 開啟MySQL主從

　?、?修改配置文件

　　找到mysql的配置文件my.cnf,并在上邊添加配置信息

log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima

　　配置解讀：

　　- `log-bin=/var/lib/mysql/mysql-bin`：設置binary log文件的存放地址和文件名，叫做mysql-bin

　　- `binlog-do-db=heima`：指定對哪個database記錄binary log events，這里記錄heima這個庫

　　以Docker容器的Mysql為例,最終配置：

[mysqld]
skip-name-resolve
character_set_server=utf8
datadir=/var/lib/mysql
server-id=1000
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima

　　5.2 設置用戶權限

　　添加一個僅用于數據同步的賬戶，出于安全考慮，這里僅提供對heima這個庫的操作權限。

　　create user canal@'%' IDENTIFIED by 'canal';
　　GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
　　FLUSH PRIVILEGES;

　　5.3 安裝Canal

　　以docker容器為例

docker run -p 11111:11111 --name canal \
-e canal.destinations=heima \
-e canal.instance.master.address=mysql: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  \
-e canal.instance.filter.regex=heima\\..* \
-d canal/canal-server:v1.1.5

　　說明:

　　- `-p 11111:11111`：這是canal的默認監(jiān)聽端口

　　- `-e canal.destinations=heima`：canal集群的名字,要和canal服務中設置的集群名字保持一致

　　- `-e canal.instance.master.address=192.168.136.133:3306`：數據庫地址和端口

　　- `-e canal.instance.dbUsername=canal`：數據庫用戶名

　　- `-e canal.instance.dbPassword=canal` ：數據庫密碼

　　- `-e canal.instance.tsdb.enable=true` ：表示是否打開tsdb開關

　　- `-e canal.instance.gtidon=false` ：表示是否是GTID模式

　　- `-e canal.instance.filter.regex=`：要監(jiān)聽的表名稱

　　表名稱監(jiān)聽支持的語法：

　　```

　　mysql 數據解析關注的表，Perl正則表達式.

　　多個正則之間以逗號(,)分隔，轉義符需要雙斜杠(\\)

　　常見例子：

　　1. 所有表：.* or .*\\..*

　　2. canal schema下所有表： canal\\..*

　　3. canal下的以canal打頭的表：canal\\.canal.*

　　4. canal schema下的一張表：canal.test1

　　5. 多個規(guī)則組合使用然后以逗號隔開：canal\\..*,mysql.test1,mysql.test2

　　```

　　5.4 Canal客戶端服務

　?、?客戶端概述

　　Canal提供了各種語言的客戶端，當Canal監(jiān)聽到binlog變化時，會通知Canal的客戶端。

　　我們可以利用Canal提供的Java客戶端，監(jiān)聽Canal通知消息。當收到變化的消息時，完成對數據的更新。

　　當然, 市面上有很多Canal客戶端, 我們以GitHub上的第三方開源的canal-starter客戶端, 并且結合SpringBoot為例, 進行演示。

　?、?引入依賴

<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

　?、?編寫配置

canal:
  destination: heima # canal的集群名字，要與安裝canal時設置的名稱一致
  server: 192.168.136.135:11111 # canal服務地址

　　④ 實體類

package com.itheima.domain;

import lombok.Data;

@Data
public class Hotel {
    private Long id;
    private String name;
    private String address;
    private Integer price;
    private Integer score;
    private String brand;
    private String city;
    private String starName;
    private String business;
    private String longitude;
    private String latitude;
    private String pic;
}

　?、?編寫監(jiān)聽器

　　> 通過實現`EntryHandler`接口編寫監(jiān)聽器，監(jiān)聽Canal消息。

　　>

　　> 注意兩點：

　　>

　　> - 實現類通過`@CanalTable("tb_hotel")`指定監(jiān)聽的表信息

　　> - EntryHandler的泛型是與表對應的實體類

　　>

package com.itheima.handle;

import com.itheima.domain.Hotel;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

@CanalTable("tb_hotel")
@Component
public class HotelHandler implements EntryHandler<Hotel> {

    /**
     * mysql中數據有新增時自動執(zhí)行
     * @param hotel 新增的數據
     */
    @Override
    public void insert(Hotel hotel) {
        //把新增數據hotel,添加到ES即可
    }

    /**
     * mysql中數據有修改時自動執(zhí)行
     * @param before 修改前的數據
     * @param after 修改后的數據
     */
    @Override
    public void update(Hotel before, Hotel after) {
        //把修改數據,更新到ES即可
    }

    /**
     * ysql中數據有刪除時自動執(zhí)行
     * @param hotel 要刪除的數據
     */
    @Override
    public void delete(Hotel hotel) {
        //把要刪除的數據hotel,從ES刪除即可
    }
}

　　6.補充

　　該模式不僅僅能實現mysql到es之間的數據同步. 它適用于任何從mysql到其他接受的數據同步。