Maxwell （mysql-to-json）初体验

来到golang学习网的大家，相信都是编程学习爱好者，希望在这里学习数据库相关编程知识。下面本篇文章就来带大家聊聊《Maxwell （mysql-to-json）初体验》，介绍一下MySQL、Kafka、同步工具，希望对大家的知识积累有所帮助，助力实战开发！

Maxwell （mysql-to-json）初体验

本篇主要讲解一下 Maxwell ， Maxwell 是一个读取 MySQL binlogs 日志然后转换成json 输出到 Kafka ，Redis ，RabbitMQ 等等 中间件中

前言

以前写过一篇关于 阿里的 canal ，它也是通过监听 mysql 的 binlogs 日志的工具，本公司目前就是使用这个，而我今天要说的是 maxwell 它是在内部自己转换为 json 格式 输出到 其他中间件

1.下载和安装Maxwell

 docker pull zendesk/maxwell

2.配置 Mysql 开起binlogs

配置 my.cnf

docker run -it --rm zendesk/maxwell bin/maxwell --user=maxwell \
    --password=maxwell --host=192.168.25.5 --producer=stdout

 insert into user(id , userName , userAge , userAddress) values (200, "johnny" , 25 , "wuxi") 

{"database":"test","table":"user","type":"insert","ts":1609312375,"xid":735,"commit":true,
 "data":{"id":200,"userName":"johnny","userAge":25,"userAddress":"wuxi"}}

4.Maxwell 输出到 kafka 模式

maxwell 官方比较推荐的 方式就是 配合 kafka 进行使用 下面先来准备 kafka 环境

4.1 启动 zookeeper

具体环境自行准备

4.2 启动kafka

4.3 启动maxwell 指定producer = kafka

    docker run -it --rm zendesk/maxwell bin/maxwell --user='maxwell' \
    --password='maxwell' --host='192.168.25.5' --producer=kafka \
    --kafka.bootstrap.servers=192.168.25.5:9092 --kafka_topic=maxwell

./kafka-console-consumer --bootstrap-server 192.168.25.5:9092 -topic maxwell

 insert into user(id , userName , userAge , userAddress) values (201, "candy" , 26 , "wuxi") 

docker run -it --rm zendesk/maxwell bin/maxwell --user='maxwell' --password='maxwell' --host='192.168.25.5' \
--producer=redis --redis_host=47.98.250.186 --redis_port=6380     --redis_type=lpush

insert into user(id , userName , userAge , userAddress) values (203, "jack" , 26 , "wuxi");
insert into user(id , userName , userAge , userAddress) values (204, "jack2" , 26 , "wuxi");
insert into user(id , userName , userAge , userAddress) values (205, "jack3" , 26 , "wuxi"); 

127.0.0.1:6379> lrange maxwell 0 -1
1) "{\"database\":\"test\",\"table\":\"user\",\"type\":\"insert\",\"ts\":1609315062,\"xid\":4462,\"commit\":true,\"data\":{\"id\":205,\"userName\":\"jack3\",\"userAge\":26,\"userAddress\":\"wuxi\"}}"
2) "{\"database\":\"test\",\"table\":\"user\",\"type\":\"insert\",\"ts\":1609315062,\"xid\":4461,\"commit\":true,\"data\":{\"id\":204,\"userName\":\"jack2\",\"userAge\":26,\"userAddress\":\"wuxi\"}}"
3) "{\"database\":\"test\",\"table\":\"user\",\"type\":\"insert\",\"ts\":1609315012,\"xid\":4322,\"commit\":true,\"data\":{\"id\":203,\"userName\":\"jack\",\"userAge\":26,\"userAddress\":\"wuxi\"}}"

6.使用 Maxwell BootStrap 初始化表

一定有 数据割接过程，或者表的 全量逻辑需求，这时候 maxwell 提供了 bootstrap 机制，可以将整个表的数据 全部发送到 producer 中

6.1 官网相关解释

bin/maxwell-bootstrap --database fooDB --table barTable

Optionally, you can include a where clause to replay part of the data.

bin/maxwell-bootstrap --database fooDB --table barTable --where "my_date >= '2017-01-07 00:00:00'"

Alternatively you can insert a row in the

mysql> insert into maxwell.bootstrap (database_name, table_name) values ('fooDB', 'barTable');

Note that if a Maxwell client_id has been set you should specify the client id.

mysql> insert into maxwell.bootstrap (database_name, table_name, client_id) values ('fooDB', 'barTable', 'custom_maxwell_client_id');

You can schedule bootstrap tasks to be run in the future by setting the started_at column. Maxwell will wait until this time to start the bootstrap.

mysql> insert into maxwell.bootstrap (database_name, table_name, client_id, started_at) 
values ('fooDB', 'barTable', 'custom

6.2 演示 使用maxwell 同步全表到kafka 中

6.2.1 准备 kafka 和 maxwell 和 consumer

docker run -it --rm zendesk/maxwell bin/maxwell-bootstrap --user maxwell  \
    --password maxwell --host=192.168.25.5  --database test --table user --client_id maxwell

--database test :指定database

--table user : 指定table = user

当上面命令执行后 可以看到 kafka-console-consumer 就能收到 database = test 库 table = user全表的数据了

{"database":"maxwell","table":"bootstrap","type":"insert","ts":1609315926,"xid":6719,"commit":true,"data":{"id":8,"database_name":"test","table_name":"user","where_clause":null,"is_complete":0,"inserted_rows":0,"total_rows":35,"created_at":null,"started_at":null,"completed_at":null,"binlog_file":null,"binlog_position":0,"client_id":"maxwell","comment":null}}

6.2.3 直接插入 bootstrap表 触发

也可以通过 对maxwell 数据的 bootstrap表插入 也能触发 ， maxwell 数据库是自动创建的

insert into maxwell.bootstrap (database_name, table_name) values ('test', 'address');

6.3 bootstrap过程中 maxwell崩溃

在进行bootstrap过程中，如果maxwell崩溃，重启时，bootstrap会完全重新开始，不管之前进行到多少，若不希望这样，可以到数据库中 maxwell 设置

is_complete

7. 扩展 Maxwell 过滤器配置

Maxwell 可以通过

--filter

exclude

include

# 仅匹配foodb数据库的tbl表和所有table_数字的表
--filter='exclude: foodb.*, include: foodb.tbl, include: foodb./table_\d+/'
# 排除所有库所有表，仅匹配db1数据库
--filter = 'exclude: *.*, include: db1.*'
# 排除含db.tbl.col列值为reject的所有更新
--filter = 'exclude: db.tbl.col = reject'
# 排除任何包含col_a列的更新
--filter = 'exclude: *.*.col_a = *'
# blacklist 黑名单，完全排除bad_db数据库，若要恢复，必须删除maxwell库
--filter = 'blacklist: bad_db.*' 

总结：

本篇主要讲解了 Maxwell 主要是干嘛的，并且介绍了 Maxwell 如何配合 Kafka 和 Redis 进行使用，最后还介绍了 Maxwell BootStrap 的操作方式，最后扩展了 Maxwell 的过滤器配置方式 。。 除了 Maxwell 还有 阿里的 Canal 你会更喜欢哪个呢 ，我比较喜欢 Maxwell 不过公司 目前在用 Canal 。。

个人博客网站 https://www.askajohnny.com 欢迎来访问！

终于介绍完啦！小伙伴们，这篇关于《Maxwell （mysql-to-json）初体验》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布数据库相关知识，快来关注吧！