Spring Boot + Canal 實現數據庫實時監控

Posted on 2022-08-22 by WalkonNet

Canal 工作原理

Canal 模擬 MySQL slave 的交互協議，偽裝自己為 MySQL slave，向 MySQL master 發送 dump 協議
MySQL master 收到 dump 請求，開始推送 binary log 給 slave（也就是 Canal）
Canal 解析 binary log 對象（原始為 byte 流）

MySQL 打開 binlog 模式

在 MySQL 配置文件 my.cnf 設置如下信息：

[mysqld]
# 打開binlog
log-bin=mysql-bin
# 選擇ROW(行)模式
binlog-format=ROW
# 配置MySQL replaction需要定義，不要和canal的slaveId重復
server_id=1

改瞭配置文件之後，重啟 MySQL，使用命令查看是否打開 binlog 模式：

SHOW VARIABLES LIKE 'log_bin';

查看 binlog 日志文件列表：SHOW BINARY LOGS;

查看當前正在寫入的 binlog 文件：SHOW MASTER STATUS; 記錄文件名 File 和 Position 值。

啟動 Canal 服務

詳細配置可以參考：

conf\example\instance.properties
 
#################################################
## mysql serverId , v1.0.26+ will autoGen 
# canal.instance.mysql.slaveId=0        //每個instance都會偽裝成一個mysql slave , 此id對於canal前端的Mysql實例而言，必須是唯一的，但是同一個Canal中相同的instance,此slaveld應該一樣
 
# 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=             //需要讀取的起始的binlog文件
canal.instance.master.position=                 //需要讀取的起始的binlog文件的偏移量
canal.instance.master.timestamp=                //需要讀取的起始的binlog的時間戳 
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                  //v1.0.25版本新增,是否開啟table meta的時間序列版本記錄功能
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb  //v1.0.25版本新增,table meta的時間序列版本的本地存儲路徑，默認為instance目錄
#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 =test         //數據庫連接時默認schema
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
canal.instance.filter.regex=.*\\..*              //mysql 數據解析關註的表，Perl正則表達式.
# table black regex
canal.instance.filter.black.regex=              //canal將會過濾那些不符合要求的table，這些table的數據將不會被解析和傳送 
#################################################

conf\canal.properties
#################################################
######### 		common argument		############# 
#################################################
canal.id= 1                       	#每個canal server實例的唯一標識
canal.ip=                           #canal server綁定的本地IP信息,如果不配置,默認選擇一個本機IP進行,
canal.port=11111                    #canal server提供socket tcp服務的端口
canal.metrics.pull.port=11112
canal.zkServers=                    #canal server鏈接zookeeper集群的鏈接信息
 
# flush data to zk
canal.zookeeper.flush.period = 1000 #canal持久化數據到zookeeper上的更新頻率,單位毫秒
canal.withoutNetty = false
 
# tcp, kafka, RocketMQ
canal.serverMode = tcp
 
# flush meta cursor/parse position to file
 
canal.file.data.dir = ${canal.conf.dir}    #canal持久化數據到file上的目錄
 
canal.file.flush.period = 1000             #canal持久化數據到file上的更新頻率,單位毫秒
 
## memory store RingBuffer size, should be Math.pow(2,n)
 
canal.instance.memory.buffer.size = 16384  #canal內存store中可緩存buffer記錄數,需要為2的指數
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024  #	內存記錄的單位大小,默認1KB,和buffer.size組合決定最終的內存使用大小
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE  #canal內存store中數據緩存模式
                                            1. ITEMSIZE : 根據buffer.size進行限制,隻限制記錄的數量
                                            2. MEMSIZE : 根據buffer.size * buffer.memunit的大小,限制緩存記錄的大小    

canal.instance.memory.rawEntry = true	
 
## 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        #心跳檢查sql
canal.instance.detecting.interval.time = 3     #心跳檢查頻率,單位秒 
canal.instance.detecting.retry.threshold = 3   #心跳檢查失敗重試次數
##非常註意：interval.time * retry.threshold值，應該參考既往DBA同學對數據庫的故障恢復時間，  
##“太短”會導致集群運行態角色“多跳”；“太長”失去瞭活性檢測的意義，導致集群的敏感度降低，Consumer斷路可能性增加。  
 
canal.instance.detecting.heartbeatHaEnable = false #心跳檢查失敗後,是否開啟自動mysql自動切換
                                                   #說明:比如心跳檢查失敗超過閥值後,如果該配置為true,canal就會自動鏈到mysql備庫獲取binlog數據	false

 
# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size =  1024   #	最大事務完整解析的長度支持超過該長度後,一個事務可能會被拆分成多次提交到canal store中,無法保證事務的完整可見性
 
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60  #canal發生mysql切換時,在新的mysql庫上查找            binlog時需要往前查找的時間,單位秒
說明:mysql主備庫可能存在解析延遲或者時鐘不統一,需要回退一段時間,保證數據不丟
 
# network config
canal.instance.network.receiveBufferSize = 16384  #網絡鏈接參數,SocketOptions.SO_RCVBUF
canal.instance.network.sendBufferSize = 16384     #網絡鏈接參數,SocketOptions.SO_SNDBUF
canal.instance.network.soTimeout = 30             #網絡鏈接參數,SocketOptions.SO_TIMEOUT
 
# binlog filter config
canal.instance.filter.druid.ddl = true          
 
canal.instance.filter.query.dcl = false           #ddl語句是否隔離發送,開啟隔離可保證每次隻返回發送一條ddl數據,不和其他dml語句混合返回.(otter ddl同步使用)
 
canal.instance.filter.query.dml = false           #是否忽略DML的query語句,比如insert/update/delete table.(mysql5.6的ROW模式可以包含statement模式的query記錄)
 
canal.instance.filter.query.ddl = false           #是否忽略DDL的query語句,比如create table/alater table/drop table/rename table/create index/drop index. (目前支持的ddl類型主要為table級別的操作,create databases/trigger/procedure暫時劃分為dcl類型)
 
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()
#canal.instance.parser.parallelThreadSize = 16
## 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
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval=24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire=360
 
# 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                       #開啟instance自動掃描
                            如果配置為true,canal.conf.dir目錄下的instance配置變化會自動觸發:
a. instance目錄新增: 觸發instance配置載入,lazy為true時則自動啟動
b. instance目錄刪除:卸載對應instance配置,如已啟動則進行關閉
c. instance.properties文件變化:reload instance配置,如已啟動自動進行重啟操作
canal.auto.scan.interval = 5                 #instance自動掃描的間隔時間,單位秒
 
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     #instance管理模式，Production級別我們要求使用spring  
canal.instance.global.lazy = false             #全局lazy模式
#canal.instance.global.manager.address = 127.0.0.1:1099  #全局的manager配置方式的鏈接信息
 
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml #全局的spring配置方式的組件文件
 
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

1）canal.deployer-1.1.5\conf\canal.properties 進行全局配置，可以修改 IP、端口號或實例

# 默認有一個 example，需要增加實例的可以配置
canal.destinations = example

2）canal.deployer-1.1.5\conf\example\instance.properties 進行局部實例配置，可以修改數據庫賬號和密碼、數據庫表名、binlog 文件名和 position 等

# 沒有改變的就沒有貼出來，註意 MySQL 的用戶名和密碼
canal.instance.master.address=192.168.58.131:3306
# username/password
canal.instance.dbUsername=test
canal.instance.dbPassword=liubihao
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=

ysql 數據解析m關註的表，Perl正則表達式.
多個正則之間以逗號(,)分隔，轉義符需要雙斜杠(\\)

常見例子：

1. 所有表：.* or .*\\..*
2. canal schema下所有表： canal\\..*
3. canal下的以canal打頭的表：canal\\.canal.*
4. canal schema下的一張表：canal.test1
5. 多個規則組合使用：canal\\..*,mysql.test1,mysql.test2 (逗號分隔)
bin/start.sh 啟動服務（Windows 系統為 bin/start.bat）

後端進行相應配置

修改 pom.xml

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

新建 CanalClient.java

package org.nwpu.atcss.util;

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 org.springframework.stereotype.Component;

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

@Component
public class CanalClient {
  private static void printEntries(List<Entry> entries) throws Exception {
    for (Entry entry : entries) {
      if (entry.getEntryType() != EntryType.ROWDATA) {
        continue;
      }

      RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());

      EventType eventType = rowChange.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 : rowChange.getRowDatasList()) {
        switch (rowChange.getEventType()) {
          case INSERT:
            System.out.println("INSERT ");
            printColumns(rowData.getAfterColumnsList());
            break;
          case UPDATE:
            System.out.println("UPDATE ");
            printColumns(rowData.getAfterColumnsList());
            break;
          case DELETE:
            System.out.println("DELETE ");
            printColumns(rowData.getBeforeColumnsList());
            break;
          default:
            break;
        }
      }
    }
  }

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

  public static void main(String[] args) throws Exception {
    // hostname, port, destination, username, password
    CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
    try {
      connector.connect();
      // 監聽的表，格式為數據庫.表名,數據庫.表名
      connector.subscribe(".*\\..*");
      connector.rollback();

      while (true) {
        Message message = connector.getWithoutAck(100); // 獲取指定數量的數據
        long batchId = message.getId();
        if (batchId == -1 || message.getEntries().isEmpty()) {
          Thread.sleep(1000);
          continue;
        }
        // System.out.println(message.getEntries());
        printEntries(message.getEntries());
        connector.ack(batchId); // 提交確認，消費成功，通知server刪除數據
        // connector.rollback(batchId);// 處理失敗, 回滾數據，後續重新獲取數據
      }
    } catch (Exception e) {
      System.out.println("Something Error.");
    } finally {
      connector.disconnect();
    }
  }
}

測試

啟動 CanalClient.java；

修改本地數據庫內容之後，控制臺成功監聽並報告更新信息。

到此這篇關於Spring Boot + Canal 實現數據庫實時監控的文章就介紹到這瞭,更多相關Spring Boot Canal 內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

Spring Boot + Canal 實現數據庫實時監控

目錄

Canal 工作原理

MySQL 打開 binlog 模式

啟動 Canal 服務

後端進行相應配置

修改 pom.xml

新建 CanalClient.java

測試

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

Canal 工作原理

MySQL 打開 binlog 模式

啟動 Canal 服務

後端進行相應配置

修改 pom.xml

新建 CanalClient.java

測試

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆