【Maxwell】01 安裝及入門

官網位址：

https://maxwells-daemon.io/      

下載下傳位址（版本發行）：

https://github.com/zendesk/maxwell/releases      

參考教程自尚矽谷視訊：

https://www.bilibili.com/video/BV1JQ4y1e7CN?spm_id_from=333.999.0.0      

什麼是Maxwell?

Zendesk 開源，用 Java 編寫的 MySQL 實時抓取軟體

功能：

實時監聽MySQL的二進制日志檔案binlog，生成JSON格式的消息

作為一個消息提供者發送給消息中間件，或者其他平台的應用程式

原理：

利用MySQL自身的主從配置特性，将自身僞裝成從庫對MySQL當成主庫進行監聽

記錄主庫的非查詢操作

版本注意：

1、版本1.30開始全面放棄使用JDK8 改用JDK11版本

2、使用JDK8，最高版本 maxwell-1.29.2.tar.gz

一、安裝Maxwell:

Linux平台 Java運作環境

1、解壓maxwell-1.29.2.tar.gz 

tar -zxvf maxwell-1.29.2.tar.gz      

2、(可選)複制解壓出來的目錄到/usr/local/中 

cp -r maxwell-1.29.2 /usr/local/      

二、配置MySQL:

1、建立Maxwell庫（庫名自定義）

CREATE DATABASE `maxwell` CHARACTER SET 'utf8mb4' COLLATE 'utf8mb4_general_ci';      

2、配置設定從庫賬号

CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL PRIVILEGES ON *.* TO 'maxwell'@'%' WITH GRANT OPTION;
FLUSH PRIVILEGES;      

3、開啟Binlog日志

[mysqld]
# 開啟binlog日志
log-bin=mysql-bin

# （可選）設定binlog日志的格式，推薦row
binlog_format= statement|mixed|row      

三、啟動Maxwell

1、方式一，直接參數式啟動

# 切換到maxwell目錄
cd /usr/local/maxwell-1.29.2

# 啟動maxwell
./bin/maxwell \
--user='maxwell' \
--password='123456' \
--host='192.168.2.225' \
--port='3308' \
--producer='stdout' \
--jdbc_options='useSSL=false&serverTimezone=Asia/Shanghai'      

參數說明：

# 主庫配置設定的從庫賬号
--user='maxwell' \ 

# 主庫配置設定的從庫賬号密碼
--password='123456' \ 

# 主庫位址
--host='192.168.2.225' \ 

# 主庫端口
--port='3308' \ 

# 消息生産模式，stdout控制台輸出，kafka
--producer='stdout' \ 

# jdbc的時區參數
--jdbc_options='useSSL=false&serverTimezone=Asia/Shanghai'      

2、加載配置檔案啟動

進入maxwell目錄：

cd /usr/local/maxwell-1.29.2      

備份配置檔案：

cp config.properties.example config.properties      

編輯關鍵參數

vim config.properties      

參數項：

# mysql login info （MySQL賬号資訊）
host=192.168.2.225
user=maxwell
password=123456
port=3308

# 訂閱資訊(生産者模式：即輸出的目标 stdout控制台， kafka kafka叢集)
producer=stdout      

3、啟動指令：

cd /usr/local/maxwell-1.29.2/
./bin/maxwell --config ./config.properties --jdbc_options='useSSL=false&serverTimezone=Asia/Shanghai'      

四、監聽效果：

檢查Maxwell是否能夠監聽？

在被監聽的主庫中執行非查詢SQL

1、插入日志

INSERT INTO `test-db`.`day_sale` (`ID`, `PRODUCT`, `CHANNEL`, `AMOUNT`, `SALE_DATE`) VALUES (NULL, '蘋果', '淘寶', 2497.0000, NOW());      

終端日志：

[root@localhost maxwell-1.29.2]# ./bin/maxwell --config ./config.properties --jdbc_options='useSSL=false&serverTimezone=Asia/Shanghai'
Using kafka version: 1.0.0
13:31:50,295 INFO  Maxwell - Starting Maxwell. maxMemory: 247332864 bufferMemoryUsage: 0.25
13:31:50,534 INFO  Maxwell - Maxwell v1.29.2 is booting (StdoutProducer), starting at Position[BinlogPosition[mysql-bin.000005:61414], lastHeartbeat=1642483666136]
13:31:50,757 INFO  MysqlSavedSchema - Restoring schema id 1 (last modified at Position[BinlogPosition[mysql-bin.000005:16191], lastHeartbeat=0])
13:31:50,931 INFO  BinlogConnectorReplicator - Setting initial binlog pos to: mysql-bin.000005:61414
13:31:50,963 INFO  BinaryLogClient - Connected to 192.168.2.225:3308 at mysql-bin.000005/61414 (sid:6379, cid:173)
13:31:50,963 INFO  BinlogConnectorReplicator - Binlog connected.
{"database":"test-db","table":"day_sale","type":"insert","ts":1642484382,"xid":65636,"commit":true,"data":{"ID":163,"PRODUCT":"蘋果","CHANNEL":"淘寶","AMOUNT":2497.0000,"SALE_DATE":"2022-01-18 13:39:42"}}      

可以看到Maxwell将會傳回一條資訊，将其格式化處理：

{
    "database": "test-db", -- 庫名
    "table": "day_sale", -- 表名
    "type": "insert", -- 執行類型：插入
    "ts": 1642484382, -- 操作時間
    "xid": 65636,  -- 操作ID？
    "commit": true, -- 送出狀态
    "data": { -- 資料資訊： {字段: 值}
        "ID": 163,
        "PRODUCT": "蘋果",
        "CHANNEL": "淘寶",
        "AMOUNT": 2497,
        "SALE_DATE": "2022-01-18 13:39:42"
    }
}      

如果是多個資料，maxwell還是根據記錄一條條列印 （說明maxwell以一條記錄為标準機關進行輸出）

{"database":"test-db","table":"day_sale","type":"insert","ts":1642484928,"xid":67070,"xoffset":0,"data":{"ID":164,"PRODUCT":"産品A","CHANNEL":"拼多多","AMOUNT":2497.0000,"SALE_DATE":"2022-01-18 13:48:48"}}
{"database":"test-db","table":"day_sale","type":"insert","ts":1642484928,"xid":67070,"xoffset":1,"data":{"ID":165,"PRODUCT":"産品B","CHANNEL":"京東","AMOUNT":2497.0000,"SALE_DATE":"2022-01-18 13:48:48"}}
{"database":"test-db","table":"day_sale","type":"insert","ts":1642484928,"xid":67070,"commit":true,"data":{"ID":166,"PRODUCT":"産品C","CHANNEL":"淘寶","AMOUNT":2497.0000,"SALE_DATE":"2022-01-18 13:48:48"}}      

2、修改記錄：

UPDATE `test-db`.`day_sale` SET `CHANNEL` = '拼多多2' WHERE `ID` = 164      

格式化資訊：

{
    "database": "test-db",
    "table": "day_sale",
    "type": "update", -- 操作類型：修改
    "ts": 1642485002, 
    "xid": 67425,
    "commit": true,
    "data": {
        "ID": 164,
        "PRODUCT": "産品A",
        "CHANNEL": "拼多多2",
        "AMOUNT": 2497,
        "SALE_DATE": "2022-01-18 13:48:48"
    },
    "old": { -- maxwell會存儲原紀錄變更的資訊
        "CHANNEL": "拼多多"
    }
}      

3、删除記錄

DELETE FROM `test-db`.`day_sale` WHERE `ID` = 165      

{
    "database": "test-db",
    "table": "day_sale",
    "type": "delete", -- 操作類型，删除
    "ts": 1642485422,
    "xid": 68499,
    "commit": true,
    "data": { -- maxwell 無論新增，删除，修改，都會對資料進行完整保留
        "ID": 165,
        "PRODUCT": "産品B",
        "CHANNEL": "京東",
        "AMOUNT": 2497,
        "SALE_DATE": "2022-01-18 13:48:48"
    }
}      

上述的操作，是終端阻塞的，要保持背景執行maxwell 就需要将消息輸出到其他中間件中去，如kafka

# 訂閱資訊(生産者模式：即輸出的目标 stdout控制台， kafka kafka叢集)
producer=kafka