订单监控_Maxwell安装

下载maxwell

• Download binary distro: https://github.com/zendesk/maxwell/releases/download/v1.22.6/maxwell-1.22.6.tar.gz

MySql环境准备

修改mysql的配置文件，开启MySQL Binlog设置

xxxxxxxxxx
[root@node1 opt]$ sudo vim /etc/my.cnf

[mysqld]
wait_timeout=86400
server_id=1
log-bin=master
binlog_format=row

重启MySQL

xxxxxxxxxx
[root@node1 opt]$ systemctl restart mysqld

查看binlog是否开启

xxxxxxxxxx
show variables like '%binlog%'

查看binlog文件

进入/var/lib/mysql目录，查看MySQL生成的binlog文件

xxxxxxxxxx
ls -l mysql | grep mysql-bin

注意：

MySQL生成的binlog文件初始大小一定是154字节，然后前缀是log-bin参数配置的，后缀是默认从.000001，然后依次递增。除了binlog文件文件以外，MySQL还会额外生产一个.index索引文件用来记录当前使用的binlog文件。

初始化元数据库

在MySQL中建立一个maxwell库用于存储Maxwell的元数据

xxxxxxxxxx
mysql> CREATE DATABASE maxwell;

设置mysql用户密码安全级别

xxxxxxxxxx
mysql> set global validate_password_policy=0;
mysql> set global validate_password_policy=LOW;
mysql> set global validate_password_length=6;

分配账号操作该数据库

xxxxxxxxxx
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%' IDENTIFIED BY '123456';

分配这个账号可以监控其他数据库的权限

xxxxxxxxxx
mysql> GRANT SELECT ,REPLICATION SLAVE , REPLICATION CLIENT ON *.* TO maxwell@'%';

刷新mysql权限

xxxxxxxxxx
mysql> flush privileges;

修改Maxwell配置文件

修改“config.properties.example”为“config.properties”并配置：

xxxxxxxxxx
# tl;dr config
log_level=info
producer=kafka
kafka_topic=my_test
kafka.bootstrap.servers=node01:9092
# mysql login info
host=localhost
user=maxwell
password=123456
jdbc_options=serverTimezone=UTC

启动Maxwell

xxxxxxxxxx
[root@node1 bin]# maxwell --config ../config.properties

向MySQL中增删改查写入数据

xxxxxxxxxx
mysql> create database mysqldb;
Query OK, 1 row affected (0.01 sec)

mysql> use mysqldb;
Database changed
mysql> create table info(id int,name varchar(255),age int);
Query OK, 0 rows affected (0.04 sec)

mysql> insert into info values (10,"xx",20);
Query OK, 1 row affected (0.01 sec)

mysql> insert into info values (12,"xx",20);
Query OK, 1 row affected (0.04 sec)

mysql> update info set age = 100 where id = 10;
Query OK, 1 row affected (0.01 sec)
Rows matched: 1  Changed: 1  Warnings: 0

mysql> delete from info where id = 10;
Query OK, 1 row affected (0.00 sec)

测试Maxwell断点续传

停止Maxwell,向MySQL中插入新的数据，重启Maxwell观察是否从上次消费到的binlog位置继续消费。

xxxxxxxxxx
#使用kill -9 xxx 命令在node01停止Maxwell
#向MySQL继续插入数据
mysql> insert into info values (200,"bbb",20);
mysql> update info set age = 30 where id = 200;
#重新在node3启动Maxwell，可以观察到Kafka中继续上次binlog位置写入数据
{"database":"mysqldb","table":"info","type":"insert","ts":1619000378,"xid":4565,"commit":true,"data"
:{"id":200,"name":"bbb","age":20}}{"database":"mysqldb","table":"info","type":"update","ts":1619000391,"xid":4566,"commit":true,"data"
:{"id":200,"name":"bbb","age":30},"old":{"age":20}}