使用canal偷取MySQL的二進制日誌

時間  2019-11-29
標籤 使用 canal 偷取 mysql 二進制 日誌 欄目 MySQL 简体版
原文   原文鏈接

前提

安裝完MySQL（我安裝的是5.7），安裝JDK（canal依賴）java

開啓MySQL的binlog

開啓binlog，而且將binlog的格式改成Row，這樣就能夠獲取到CURD的二進制內容。配置/etc/my.cnf，在[mysqld]增長mysql

log-bin=mysql-bin #添加這一行就ok
binlog-format=ROW #選擇row模式
server_id=1 # 惟一，不能和其餘集羣MySQL的server_id同樣
複製代碼

驗證binlog是否開啓

登陸MySQL，使用命令：git

show variables like 'log_%'；
複製代碼

若 log_bin顯示爲 on ，則說明開啓。github

給canal分配MySQL的帳號

給canal分配一個MySQL的帳號，方便canal偷取MySQL的binlog。spring

CREATE USER canal IDENTIFIED BY 'canal';
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
複製代碼

查看是否給canal帳號分配權限sql

show grants for 'canal' 
複製代碼

下載解壓canal

地址：github.com/alibaba/can… ，目前穩定版是 v1.1.0，下載 canal.deployer-1.1.0.tar.gz。解壓到 canal目錄下（沒有該目錄 就新建）shell

注：canal 是純Java寫的，全部須要依賴JDK環境，我這邊使用的是：1.8.0_65-b17數據庫

# 下載
wget https://github.com/alibaba/canal/releases/download/canal-1.1.0/canal.deployer-1.1.0.tar.gz
# 建立canal目錄
mkdir canal
# 解壓
tar -zxvf https://github.com/alibaba/canal/releases/download/canal-1.1.0/canal.deployer-1.1.0.tar.gz 
複製代碼

canal和instance配置文件

一個canal裏面可能會有多個instance，也就說一個instance能夠監控一個mysql實例，多個instance也就能夠對應多臺服務器的mysql實例。也就是一個canal就能夠監控分庫分表下的多機器MySQL。緩存

（1）canal.properties

canal/config 中的canal.properties文件，是全局性的canal服務器配置 ，修改內容以下：bash

#################################################
######### 		common argument		############# 
#################################################
# id惟一，不可與mysql的server_id重複
canal.id= 2		
canal.ip=
canal.port=11111
canal.metrics.pull.port=11112
canal.zkServers=
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024 
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE

## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false

# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size =  1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60

# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30

# binlog filter config
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false

# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED 
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

# binlog ddl isolation
canal.instance.get.ddl.isolation = false

# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
# parallelThreadSize默認是註釋掉的，原值爲16，由於canal裝在本地VM上，分配了1個CPU，致使報錯，改成1
canal.instance.parser.parallelThreadSize = 1
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256

# table meta tsdb info
canal.instance.tsdb.enable=true
canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername=canal
canal.instance.tsdb.dbPassword=canal

# rds oss binlog account
canal.instance.rds.accesskey =
canal.instance.rds.secretkey =

#################################################
######### 		destinations		############# 
#################################################
canal.destinations= example
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5

canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode = spring 
canal.instance.global.lazy = false
#canal.instance.global.manager.address = 127.0.0.1:1099
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

# position info，須要改爲本身的數據庫信息  
canal.instance.master.address = 127.0.0.1:3306   
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.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 = .*\\..*

複製代碼
（2）instance.properties

位於 canal/example/instance.properties，是具體的某個instances實例的配置，未涉及到的配置都會從canal.properties上繼承，內容以下：

#################################################
## mysql serverId , v1.0.26+ will autoGen 
# canal.instance.mysql.slaveId=0

# enable gtid use true/false
canal.instance.gtidon=false

# position info address修改成本身的mysql地址
canal.instance.master.address=192.168.204.128:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=

# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=

# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal

#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position = 
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=

# username/password  修改成在mysql中給canal同步數據的帳號 密碼
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=
#################################################
複製代碼

建立test數據庫

查看MySQL上是否有test數據庫，沒有則建立

開啓canal

進入canal/bin，執行：./startup.sh。

使用 ps -ef|grep canal 驗證是否開啓。

Java client代碼

建立SpringBoot工程，引入依賴：

<dependency>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.client</artifactId>
     <version>1.1.0</version>
 </dependency>
複製代碼

建立TestCanal類：

package com.xbq.canal.test;

import java.awt.Event;
import java.net.InetSocketAddress;
import java.util.List;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;

/** * @Auther: xbq * @Date: 2018/9/11 19:16 * @Description: */
public class TestCanal {

    public static void main(String[] args) throws InterruptedException {
        // 第一步：與canal進行鏈接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.204.128", 11111),
                "example", "", "");
        connector.connect();
        // 第二步：開啓訂閱
        connector.subscribe();
        // 第三步：循環訂閱
        while (true) {
            try {
                // 每次讀取 1000 條
                Message message = connector.getWithoutAck(1000);
                long batchID = message.getId();
                int size = message.getEntries().size();
                if (batchID == -1 || size == 0) {
                    System.out.println("當前暫時沒有數據");
                    Thread.sleep(1000);
                } else {
                    System.out.println("-------------------------- 有數據啦 -----------------------");
                    PrintEntry(message.getEntries());
                }
                // position id ack （方便處理下一條）
                connector.ack(batchID);
            } catch (Exception e) {
                // TODO: handle exception
            } finally {
                Thread.sleep(1000);
            }
        }
    }

    /** * 獲取每條打印的記錄 * @param entrys */
    public static void PrintEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            // 第一步：拆解entry 實體
            Header header = entry.getHeader();
            EntryType entryType = entry.getEntryType();

            // 第二步： 若是當前是RowData，那就是我須要的數據
            if (entryType == EntryType.ROWDATA) {
                String tableName = header.getTableName();
                String schemaName = header.getSchemaName();
                RowChange rowChange = null;
                try {
                    rowChange = RowChange.parseFrom(entry.getStoreValue());
                } catch (InvalidProtocolBufferException e) {
                    e.printStackTrace();
                }
                EventType eventType = rowChange.getEventType();
                System.out.println(String.format("當前正在操做 %s.%s， Action= %s", schemaName, tableName, eventType));

                // 若是是‘查詢’ 或者 是 ‘DDL’ 操做，那麼sql直接打出來
                if (eventType == EventType.QUERY || rowChange.getIsDdl()) {
                    System.out.println("rowchange sql ----->" + rowChange.getSql());
                    return;
                }
                // 第三步：追蹤到 columns 級別
                rowChange.getRowDatasList().forEach((rowData) -> {
                    // 獲取更新以前的column狀況
                    List<Column> beforeColumns = rowData.getBeforeColumnsList();
                    // 獲取更新以後的 column 狀況
                    List<Column> afterColumns = rowData.getAfterColumnsList();
                    // 當前執行的是 刪除操做
                    if (eventType == EventType.DELETE) {
                        PrintColumn(beforeColumns);
                    }
                    // 當前執行的是 插入操做
                    if (eventType == EventType.INSERT) {
                        PrintColumn(afterColumns);
                    }
                    // 當前執行的是 更新操做
                    if (eventType == EventType.UPDATE) {
                        PrintColumn(afterColumns);
                    }
                });
            }
        }
    }

    /** * 每一個row上面的每個column 的更改狀況 * @param columns */
    public static void PrintColumn(List<Column> columns) {
        columns.forEach((column) -> {
            String columnName = column.getName();
            String columnValue = column.getValue();
            String columnType = column.getMysqlType();
            // 判斷 該字段是否更新
            boolean isUpdated = column.getUpdated();
            System.out.println(String.format("columnName=%s, columnValue=%s, columnType=%s, isUpdated=%s", columnName,
                    columnValue, columnType, isUpdated));
        });
    }
}
複製代碼

運行此類。在MySQL test數據庫中建立student表，對其進行增刪改，能夠發現控制檯上打印：有數據庫啦……

參考

[緩存一致性和跨服務器查詢的數據異構解決方案canal


歡迎關注個人公衆號，第一時間接收最新文章~ 搜索公衆號： 碼咖 或者 掃描下方二維碼：

img
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程