關於使用Binlog和canal來對MySQL的數據寫入進行監控

先說下Binlog和canal是什么吧。

1、Binlog是mysql數據庫的操作日志，當有發生增刪改查操作時，就會在data目錄下生成一個log文件，形如mysql-bin.000001，mysql-bin.000002等格式

2、canal是阿里巴巴旗下的一款開源項目，純Java開發。基於數據庫增量日志解析，提供增量數據訂閱&消費，目前主要支持了MySQL（也支持mariaDB）；

3、canal起源：早期，阿里巴巴B2B公司因為存在杭州和美國雙機房部署，存在跨機房同步的業務需求。不過早期的數據庫同步業務，主要是基於trigger的方式獲取增量變更，不過從2010年開始，阿里系公司開始逐步的嘗試基於數據庫的日志解析，獲取增量變更進行同步，由此衍生出了增量訂閱&消費的業務，從此開啟了一段新紀元。

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

（1）數據庫鏡像

（2）數據庫實時備份

（3）多級索引 (賣家和買家各自分庫索引)

（4）search build

（5）業務cache刷新

（6）價格變化等重要業務消息

二、工作原理：

1、mysql主備復制實現：

從上層來看，復制分成三步：

（1）master將改變記錄到二進制日志(binary log)中（這些記錄叫做二進制日志事件，binary log events，可以通過show binlog events進行查看）；

（2）slave將master的binary log events拷貝到它的中繼日志(relay log)；

（3）slave重做中繼日志中的事件，將改變反映它自己的數據。

2、canal的工作原理：

原理相對比較簡單：

（1）canal模擬mysql slave的交互協議，偽裝自己為mysql slave，向mysql master發送dump協議

（2）mysql master收到dump請求，開始推送binary log給slave(也就是canal)

（3）canal解析binary log對象(原始為byte流)

三、主要配置：

1、mysql默認是沒有開啟Binlog，不妨查看下本地mysql是否開啟，可執行：

SHOW VARIABLES LIKE 'log_%';

如下圖，是我本地mysql，“on”代表已經開啟，“off”代表關閉

2、如何開啟Binlog：

（1）先進入路徑為C:\ProgramData\MySQL\MySQL Server 5.6的文件夾下(如果沒有，可能是沒有將隱藏的文件夾顯示)，而不是這個路徑C:\Program Files\MySQL\MySQL Server 5.6

找到my.ini文件，在打開之前需要先將mysql服務停止，之后在my.ini配置文件中添加以下內容：

#添加這一行就ok  
log-bin=mysql-bin 
#選擇row模式 
binlog-format=ROW  
#配置mysql replaction需要定義，不能和canal的slaveId重復  
server_id=1 

#指定生成log的數據庫
binlog_do_db=springboot_test1（這是指定需要生成log的數據庫，如果刪除這句，則表示所有數據庫都需要生成log）

log-output=FILE
general-log=1  （只需要將0更改為1即可）
general_log_file="MYUNYU.log"
slow-query-log=1
slow_query_log_file="MYUNYU-slow.log"
long_query_time=10

配置好之后，再啟動mysql服務，執行查看binlog是否開啟，如果還沒開啟，那就可能是配置出了問題或者mysql版本的問題，這里不詳細說

（2）從節點通過一個專門的賬號連接主節點，這個賬號需要擁有全局的 REPLICATION 權限。我們可以使用GRANT 命令創建這樣的賬號（需要先選中mysql系統庫）：

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

再查詢mysql系統庫中的user表是否存在canal用戶：

3、配置canal：

首先先下載canal服務端代碼canal.deployer-1.1.1.tar.gz（https://github.com/alibaba/canal/releases），解壓之后，配置文件在conf文件夾下，

進入路徑為C:\...\canal\canal.deployer-1.1.1\conf\example的文件夾，打開配置文件instance.properties,詳細配置如下：

#################################################
## mysql serverId , v1.0.26+ will autoGen 
　　#slaveId不能與my.ini配置文件的server_id一致
canal.instance.mysql.slaveId=1234

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

# 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.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/springboot_test1
#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
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
　　#這里配置監控的數據庫名
canal.instance.defaultDatabaseName =springboot_test1
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=
#################################################

再進入路徑為C:\...\canal\canal.deployer-1.1.1\bin的文件夾，雙擊打開startup.bat，如果顯示有以下內容，則表示配置成功，canal服務器端啟動ok：

4、客戶端代碼測試：

（1）首先創建一個空的maven項目，在pom文件中引入canal客戶端的依賴：

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

（2）創建一個類進行測試：

package com.xxx;

import java.net.InetSocketAddress;
import java.util.List;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
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.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.client.*;

public class CanalClient {

    public static void main(String args[]) {
        // 創建鏈接            instanceA
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                11111), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalEntryCount = 1200;
            while (emptyCount < totalEntryCount) {
                Message message = connector.getWithoutAck(batchSize); // 獲取指定數量的數據
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    emptyCount = 0;
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交確認
            }
            System.out.println("empty too many times, exit");
        }catch (Exception e){
            //connector.rollback(batchId); // 處理失敗, 回滾數據
        }
        finally {
            connector.disconnect();
        }
    }

    private static void printEntry( List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());

                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private static void printColumn( List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

}

運行CanalClient的main方法，如果看到控制台出現以下內容則代表連接成功：

再到mysql中創建數據庫springboot_test,springboot_test1,springboot_test2,再在這三個庫中分別創建student表，sql語句為：

CREATE TABLE `student` (
`ID` int(11) NOT NULL AUTO_INCREMENT,
`NAME` varchar(20) NOT NULL,
`CLASS_NAME` varchar(30) NOT NULL,
`CREATE_DATE` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`UPDATE_DATE` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`ID`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC;

執行插入語句：

INSERT INTO student(NAME , class_name )VALUES('student1' , 'class1');

sql執行完之后再看idea中的控制台，如果出現下面內容打印則表示可以監控數據庫的寫入等操作

到此，使用Binlog和canal來對MySQL的數據寫入進行監控的操作實現完畢！