使用Maxwell實時同步mysql數據

   接上一篇Mysql主從複製的搭建-基於Docker，本文介紹用Maxwell假裝成mysql從節點，接收binlog的使用。

• Maxwell簡介

    maxwell是由java編寫的守護進程，能夠實時讀取mysql binlog並將行更新以JSON格式寫入kafka、rabbitMq、redis等中，  這樣有了mysql增量數據流，使用場景就不少了，好比：實時同步數據到緩存，同步數據到ElasticSearch，數據遷移等等。

  maxwell官網：http://maxwells-daemon.io

  maxwell源代碼：https://github.com/zendesk/maxwell

•   Maxwell的配置與使用

    1.下載Maxwell安裝包

root@xxx maxwell]# pwd
/usr/local/maxwell
[root@xxx maxwell]# wget https://github.com/zendesk/maxwell/releases/download/v1.19.5/maxwell-1.19.5.tar.gz
[root@xxx maxwell]# tar zxvf maxwell-1.19.5.tar.gz 
[root@xxx maxwell]# cd maxwell-1.19.5

   2.配置mysql，打開mysql binlog日誌

[root@xxx mysql]# vi /usr/local/mysql/my.cnf 


[mysqld]
log-bin=mysql-bin #添加這一行就
binlog-format=ROW #選擇row模式
server_id=1 #隨機指定一個不能和其餘集羣中機器重名的字符串，若是隻有一臺機器，那就能夠隨便指定了

   重啓mysql服務，登錄mysql，查看binlog日誌模式

mysql> show variables like '%log_bin%'
+---------------------------------+-------------------------------+
| Variable_name                   | Value                         |
+---------------------------------+-------------------------------+
| log_bin                         | ON                            |
| log_bin_basename                | /data/mysqldb/mysql-bin       |
| log_bin_index                   | /data/mysqldb/mysql-bin.index |
| log_bin_trust_function_creators | OFF                           |
| log_bin_use_v1_row_events       | OFF                           |
| sql_log_bin                     | ON                            |
+---------------------------------+-------------------------------+
6 rows in set (0.11 sec)

  Maxwell須要在schema_database選項指定的數據庫中存儲狀態的權限（默認庫名稱爲maxwell），因此須要提早給權限：

#建立一個有同步數據的用戶yhrepl
mysql> create user 'yhrepl'@'*' identified by 'scgaopan'; 
Query OK, 0 rows affected (0.10 sec)

#此用戶yhrepl要有對須要同步的數據庫表有操做權限
mysql> grant all privileges on test.* to 'yhrepl'@'%' identified by 'scgaopan'; 
Query OK, 0 rows affected (0.13 sec)

#給yhrepl有同步數據的權限
mysql> grant select,replication client,replication slave on *.* to 'yhrepl'@'%' identified by 'scgaopan';
Query OK, 0 rows affected (0.10 sec)
# Maxwell須要在schema_database選項指定的數據庫中存儲狀態的權限（默認庫名稱爲maxwell）
mysql> grant all privileges on maxwell.* to 'yhrepl'@'%' identified by 'scgaopan';
Query OK, 0 rows affected (0.09 sec)

 3.啓動Maxwell，主要介紹數據寫入rabbitmq的實戰：

[root@xxx maxwell-1.19.5]# vi /usr/local/maxwell/maxwell-1.19.5/config.properties
#日誌級別
log_level=DEBUG

producer=rabbitmq
daemon=true

#監控的數據庫
host=47.105.110.xxx
user=yhrepl
password=scgaopan

output_nulls=true
jdbc_options=autoReconnet=true

#監控數據庫中的哪些表
filter=exclude: *.*,include: test.AA
replica_server_id=64
client_id=test-id

#metrics_type=http
#metrics_slf4j_interval=60
#http_port=8111
#http_diagnostic=true # default false


#rabbitmq
rabbitmq_host=47.105.110.xxx
rabbitmq_port=5672
rabbitmq_user=guest
rabbitmq_pass=guest
rabbitmq_virtual_host=/
rabbitmq_exchange=maxwell
rabbitmq_exchange_type=topic
rabbitmq_exchange_durable=false
rabbitmq_exchange_autodelete=false
rabbitmq_routing_key_template=%db%.%table%
rabbitmq_message_persistent=false
rabbitmq_declare_exchange=true

啓動Maxwell：

[root@xxx maxwell-1.19.5]# ./bin/maxwell
#能夠後臺啓動
[root@xxx maxwell-1.19.5]# nohub ./bin/maxwell &

rabbitmq的操做，啓動maxwell後就有一個maxwell的exchage生成

但對應的queue和exchange和queue的綁定須要用戶本身去實現

新建一個maxwell-test的queue:

 

把queue與exchange進行綁定：

注意，這裏的Routing key 是區分大小寫的

在數據庫中修改一條記錄，能夠看到maxwell-test隊列裏面有一第記錄了。