当前位置：首页 > news >正文

Maxwell监听mysql的binlog日志变化写入kafka消费者

news 2025/8/3 21:44:24

一. 环境： maxwell:v1.29.2 (从1.30开始maxwell停止了对java8的使用，改为为11)

maxwell1.29.2这个版本对mysql8.0以后的缺少utf8mb3字符的解码问题，需要对原码中加上一个部分内容：具体也给大家做了总结：

关于v1.29.2 版本的Maxwell存在于mysql8.0后版本部分源码字符集处理确实问题-CSDN博客

二. 程序这里还是那一个kafka模拟器来实时向mysql写数据做测试：

maxwell可以成功将监听到的binlog日志信息写入到kafka的主题中去消费

三. 关于maxwell的配置以及启动方式：

1. 配置mysql的my.cnf配置文件开启binlog日志

 sudo  vim  /etc/my.cnf在[mysqld]模块下添加一下内容[mysqld]
server_id=1
log-bin=mysql-bin
binlog_format=row#binlog-do-db=test_maxwell然后并重启 Mysql 服务
sudo  systemctl  restart  mysqld登录 mysql 并查看是否修改完成 使用：
mysql>   show  variables  like  '%binlog%';
查看下列属性binlog_format	|   ROW

2.查看mysql生成的binlog日志文件:正常会生成二个文件一个初始化文件，一个索引记录文件

cd  /var/lib/mysqlsudo  ls  -l |grep binlog
总用量  188500
-rw-r-----.   1   mysql   mysql	154   11 月  17   16:30   mysql-
bin.000001
-rw-r-----.   1   mysql   mysql	19   11 月  17   16:30   mysql-
bin.index

3.初始化Maxwell元数据库

（1）建立一个 maxwell 库用于存储 Maxwell 的元数据
mysql>   CREATE   DATABASE   maxwell;
（2）设置 mysql 用户密码安全级别感觉8.1后也不管用啊
mysql> set   global   validate_password_length=4;
mysql> set   global   validate_password_policy=0;
（3）创建一个maxwell账号可以操作该数据库 :user=maxwell ,PW：123456
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%' IDENTIFIED BY '123456';
（4）分配这个账号可以监控其他数据库的权限，查看其对应的binlog日志
mysql> GRANT SELECT ,REPLICATION SLAVE , REPLICATION CLIENT ON
*.*  TO  maxwell@'%';
（5）最后记得刷新权限即可
mysql>flush   privileges;

4.启动maxwell的方式有二种：

4.1 .纯指令式

4.1.1 ：打印在console

bin/maxwell 
--user='maxwell' 
--password='123456' 
--host='hadoop102' 
--producer=stdout

4.1.2 ： binlog消费到kafka的topic中

bin/maxwell 
--user='maxwell'
--password='123456!'
--host='localhost'
--producer=kafka 
--kafka.bootstrap.servers=hadoop103:9092,hadoop104:9092
--kafka_topic=maxwelltest

4.2 指定config文件式

vim  config.properties#   tl;dr   config
log_level=info
producer=kafka#   mysql   login   info
host=hadoop102
user=maxwell
password=123456#	***   kafka   ***
#   list   of   kafka   brokerskafka.bootstrap.servers=hosta:9092,hostb:9092
kafka_topic=maxwell

这种config式的配置之后在使用指令式的时候不能再加host参数和指定value了会报错

5.开启kafka消费者来消费binlog二进制数据

kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_name# 如果要kafka消费的数据直接写入hdfs：kafka-console-consumer.sh --bootstrap-server hadoop103:9092,hadoop104:9092 --topic topic-name | hadoop fs -put - /dataset/lixianData/text.txt  # 这里写入到这个txt文件中，提前建好

查看全文

http://www.lryc.cn/news/324330.html