簡介

Maxwell是由Java語言編寫,Zendesk開源的binlog解析同步工具。可通過簡單配置，將binlog解析並以json的格式同步到如file,kafka,redis,RabbitMQ等系統中。也可自定義輸出。相比Canal,Maxwell相當於Canal Server+Canal Client。

安裝

配置MySQL

MySQL 開啟Binlog

#開啟binlog
#修改my.cnf配置檔案 增加如下內容
[[email protected] /root]# vim /etc/my.cnf

[mysqld]
#binlog檔案儲存目錄及binlog檔名字首
 

#binlog檔案儲存目錄: /var/lib/mysql/
#binlog檔名字首: mysql-binlog
#mysql向檔名字首新增數字字尾來按順序建立二進位制日誌檔案 如mysql-binlog.000006 mysql-binlog.000007
log-bin=/var/lib/mysql/mysql-binlog
#選擇基於行的日誌記錄方式
binlog-format=ROW
#伺服器 id
#binlog資料中包含server_id,標識該資料是由那個server同步過來的
server_id=1

MySQL 配置許可權

CREATE USER 'maxwell_sync'@'%' IDENTIFIED BY
 
 'maxwell_sync_1';
-- Maxwell需要在待同步的庫上建立schema_database庫,將狀態儲存在`schema_database`選項指定的資料庫中(預設為`maxwell`)
GRANT ALL on maxwell.* to 'maxwell_sync'@'%';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'maxwell_sync'@'%';
FLUSH PRIVILEGES;

MySQL 建庫建表

create database test_maxwell;
use test_maxwell;
create
 
 table if not exists `user_info`(
   `userid` int,
   `name` varchar(100),
   `age` int
)engine=innodb default charset=utf8;

配置Maxwell

下載解壓

[root@node2 /data/software]# wget https://github.com/zendesk/maxwell/releases/download/v1.17.1/maxwell-1.17.1.tar.gz

[root@node2 /data/software]# tar -zxvf maxwell-1.17.1.tar.gz

解析binlog並同步至kafka並配置監控

啟動maxwell

#輸入來源於mysql binlog 
#輸出到kafka
#配置說明
#1)kafka_topic 
#可配置成如 namespace_%{database}_%{table} %{database} 和 %{table}會被替換成真正的值
#2)kafka_version 
#注意和kafka版本匹配。
#3)額外配置 
#kafka.acks、kafka.compression.type、kafka.retries
#4)filter
#可排除庫、表、過濾掉某些行。也可用一段js靈活處理資料 
#如 exclude: test_maxwell.user_info.userid = 1 排除test_maxwell庫user_info表userid值為1的行
#5)monitor
#可配置的監控方式jmx、http等
#http_bind_address 監控繫結的IP
#http_port 監控繫結的Port
#http_path_prefix http請求字首

[[email protected] /data/software/maxwell-1.17.1]# bin/maxwell \
--host='localhost' \
--port=3306 \
--user='maxwell_sync' \
--password='maxwell_sync_1' \
--filter='exclude: *.*,include:test_maxwell.user_info,exclude: test_maxwell.user_info.userid = 1' \
--producer=kafka \
--kafka_version='0.11.0.1' \
--kafka.bootstrap.servers='node1:6667,node2:6667,node3:6667' \
--kafka_topic=qaTopic \
--metrics_type=http \
--metrics_jvm=true \
--http_bind_address=node2 \
--http_port=8090 \
--http_path_prefix=db_test_maxwell

#輸出到控制檯用如下配置

[[email protected] /data/software/maxwell-1.17.1]# bin/maxwell \
--host='localhost' \
--port=3306 \
--user='maxwell_sync' \
--password='maxwell_sync_1' \
--producer=stdout

kafka消費者客戶端

[[email protected] /usr/hdp/2.6.4.0-91/kafka]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.113.101:6667 --topic qaTopic

解析Insert

#sql insert 3條資料
mysql> insert into user_info(userid,name,age) values (1,'name1',10),(2,'name2',20),(3,'name3',30);

#kafka-console-consumer結果
#userid=1的資料被過濾掉了
{"database":"test_maxwell","table":"user_info","type":"insert","ts":1533857131,"xid":10571,"xoffset":0,"data":{"userid":2,"name":"name2","age":20}}
{"database":"test_maxwell","table":"user_info","type":"insert","ts":1533857131,"xid":10571,"commit":true,"data":{"userid":3,"name":"name3","age":30}}

解析Delete

#sql delete
mysql> delete from user_info where userid=2;

#kafka-console-consumer結果
{"database":"test_maxwell","table":"user_info","type":"delete","ts":1533857183,"xid":10585,"commit":true,"data":{"userid":2,"name":"name2","age":20}}

解析Update

#sql update
mysql> update user_info set name='name3',age=23 where userid=3;

#maxwell解析結果
{"database":"test_maxwell","table":"user_info","type":"update","ts":1533857219,"xid":10595,"commit":true,"data":{"userid":3,"name":"name3","age":23},"old":{"age":30}}

檢視監控

訊息處理速度與JVM

成功傳送到Kafka的訊息數、傳送失敗的訊息數
已從binlog處理的行數、消費binlog速度、jvm狀態
http://node2:8090/db_test_maxwell/metrics

返回:
counters: {
MaxwellMetrics.messages.failed: {
count: 0
},
MaxwellMetrics.messages.succeeded: {
count: 0
},
MaxwellMetrics.row.count: {
count: 84
}
}
......

maxwell健康狀態

http://node2:8090/db_test_maxwell/healthcheck

返回:
{
MaxwellHealth: {
healthy: true
}
}

ping

http://node2:8090/db_test_maxwell/ping

能ping通返回字串pong

Maxwell優缺點

優點

(1) 相比較canal,配置簡單,開箱即用。
(2) 可自定義傳送目的地(java 繼承類,實現方法),資料處理靈活(js)。
(3) 自帶多種監控。

缺點

(1) 需要在待同步的業務庫上建schema_database庫(預設maxwell),用於存放元資料,如binlog消費偏移量。但按maxwell的意思,這並不是缺點。
(2) 不支援HA。而canal可通過zookeeper實現canal server和canal client的HA,實現failover。