maxwell同步mysql到kafka(一个服务器启动多个)

创建mysql同步用户

CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%'; 

开启mysql binlog

a.修改 /etc/my.cnf 配置

log-bin=mysql-bin  # 开启binlog
binlog-format=ROW  # 设置Binary Log记录方式为Row
server_id=1 # 记住id 后续开发会使用
# 指定binlog日志文件的名字为mysql-bin,以及其存储路径
# 如果没有对log-bin指定log文件,默认在 /var/lib/mysql目录下以mysqld-bin.00000X等作为名称。
# 而 mysqld-bin.index则记录了所有的log的文件名称
# 使用时则使用mysqlbinlog /var/lib/mysql|grep "*****"等来追踪database的操作。
log-bin=/var/lib/mysql/mysql-bin   
datadir=/var/lib/mysql
socket=/var/lib/mysql/mysql.sock

b.重启mysql
service mysqld restart
c.查看开启状态
输入 show variables like ‘log_bin’; 查看binlog开启状态。如下图所示。
输入 show variables like ‘binlog_format’; 查看Binary Log记录方式。如下图所示。

mysql> show variables like 'log_%';
+---------------------------------+-------------+
| Variable_name | Value |
+---------------------------------+-------------+
| log_bin | ON |
| log_bin_trust_function_creators | OFF |
| log_error | .\mysql.err |
| log_queries_not_using_indexes | OFF |
| log_slave_updates | OFF |
| log_slow_queries | ON |
| log_warnings | 1 |
+---------------------------------+-------------+
没有开启log_bin的值是OFF,开启之后是ON
mysql> SHOW VARIABLES LIKE 'character%';
+--------------------------+----------------------------+
| Variable_name            | Value                      |
+--------------------------+----------------------------+
| character_set_client     | utf8                       |
| character_set_connection | utf8                       |
| character_set_database   | latin1                     |
| character_set_filesystem | binary                     |
| character_set_results    | utf8                       |
| character_set_server     | latin1                     |
| character_set_system     | utf8                       |
| character_sets_dir       | /usr/share/mysql/charsets/ |
+--------------------------+----------------------------+

安装maxwell

下载

从 v1.30.0 开始,Maxwell 不再支持 JDK1.8
使用最后一个支持 JDK1.8 版本的 Maxwell v1.29.2 进行部署
wget https://github.com/zendesk/maxwell/releases/download/v1.29.2/maxwell-1.29.2.tar.gz

安装

解压即可 tar -zxvf maxwell.tar.gz

配置

vim config_1.properties

server_id=1
client_id=city_ct_63 #用于启动多个maxwell
replica_server_id=2 #用于启动多个maxwell
user=maxwell
password=123456
host=192.168.0.XX
producer=kafka
kafka.bootstrap.servers=192.168.0.XX:9092
kafka_topic=city_mysql_kafka_cdc
jdbc_options=serverTimezone=UTC
filter=exclude:*.*,include:test.mj_qyxx,include:test.mj_sbxx

vim config_2.properties

server_id=1
client_id=province_ct_63 #用于启动多个maxwell
replica_server_id=1 #用于启动多个maxwell
user=maxwell
password=123456
host=192.168.0.1x
producer=kafka
kafka.bootstrap.servers=192.168.0.xx:9092
kafka_topic=province_mysql_kafka_cdc
jdbc_options=serverTimezone=UTC
filter=exclude:*.*,include:test2.xxx

启动

bin/maxwell --config city_config.properties --daemon
bin/maxwell --config province_config.properties --daemon

验证启动进程

ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l

测试数据库全量同步

maxwell-bootstrap --database finance_result --table industry --config $MAXWELL_HOME/config.properties

其中发送过来的数据第一行及最后一行数据是标识 Maxwell 历史数据同步的,不携带任何数据。

maxwell -> kafka: 
{"database": "finance_result","table": "industry","type": "bootstrap-start","ts": 1694748250,"data": {}
}{"database": "finance_result","table": "industry","type": "bootstrap-insert","ts": 1694748250,"data": {"id": 1,"create_time": "2022-08-19 00:00:00.000000","update_time": "2022-08-19 00:00:00.000000","industry_level": 1,"industry_name": "工程建设","superior_industry_id": null}
} {"database": "finance_result","table": "industry","type": "bootstrap-insert","ts": 1694748250,"data": {"id": 2,"create_time": "2022-08-19 00:00:00.000000","update_time": "2022-08-19 00:00:00.000000","industry_level": 1,"industry_name": "轻工","superior_industry_id": null}
} {"database": "finance_result","table": "industry","type": "bootstrap-insert","ts": 1694748250,"data": {"id": 3,"create_time": "2022-08-19 00:00:00.000000","update_time": "2022-08-19 00:00:00.000000","industry_level": 2,"industry_name": "土木","superior_industry_id": 1}
}
......{"database": "finance_result","table": "industry","type": "bootstrap-complete","ts": 1694748250,"data": {}
}

测试数据库增量同步

参数说明

输出JSON字符串的格式
● data 最新的数据,修改后的数据
● old 旧数据,修改前的数据
● type 操作类型,有insert, update, delete, database-create, database-alter, database-drop, table-create, table-alter, table-drop,bootstrap-insert,int(未知类型)
● xid 事务id
● commit 同一个xid代表同一个事务,事务的最后一条语句会有commit,可以利用这个重现事务
● server_id
● thread_id
● 运行程序时添加参数–output_ddl,可以捕捉到ddl语句
● datetime列会输出为"YYYY-MM-DD hh:mm:ss",如果遇到"0000-00-00 00:00:00"会原样输出
● maxwell支持多种编码,但仅输出utf8编码
● maxwell的TIMESTAMP总是作为UTC处理,如果要调整为自己的时区,需要在后端逻辑上进行处理

创建mysql同步用户

CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%'; 

开启mysql binlog

a.修改 /etc/my.cnf 配置

log-bin=mysql-bin  # 开启binlog
binlog-format=ROW  # 设置Binary Log记录方式为Row
server_id=1 # 记住id 后续开发会使用
# 指定binlog日志文件的名字为mysql-bin,以及其存储路径
# 如果没有对log-bin指定log文件,默认在 /var/lib/mysql目录下以mysqld-bin.00000X等作为名称。
# 而 mysqld-bin.index则记录了所有的log的文件名称
# 使用时则使用mysqlbinlog /var/lib/mysql|grep "*****"等来追踪database的操作。
log-bin=/var/lib/mysql/mysql-bin   
datadir=/var/lib/mysql
socket=/var/lib/mysql/mysql.sock

b.重启mysql
service mysqld restart
c.查看开启状态
输入 show variables like ‘log_bin’; 查看binlog开启状态。如下图所示。
输入 show variables like ‘binlog_format’; 查看Binary Log记录方式。如下图所示。

mysql> show variables like 'log_%';
+---------------------------------+-------------+
| Variable_name | Value |
+---------------------------------+-------------+
| log_bin | ON |
| log_bin_trust_function_creators | OFF |
| log_error | .\mysql.err |
| log_queries_not_using_indexes | OFF |
| log_slave_updates | OFF |
| log_slow_queries | ON |
| log_warnings | 1 |
+---------------------------------+-------------+
没有开启log_bin的值是OFF,开启之后是ON
mysql> SHOW VARIABLES LIKE 'character%';
+--------------------------+----------------------------+
| Variable_name            | Value                      |
+--------------------------+----------------------------+
| character_set_client     | utf8                       |
| character_set_connection | utf8                       |
| character_set_database   | latin1                     |
| character_set_filesystem | binary                     |
| character_set_results    | utf8                       |
| character_set_server     | latin1                     |
| character_set_system     | utf8                       |
| character_sets_dir       | /usr/share/mysql/charsets/ |
+--------------------------+----------------------------+

安装maxwell

下载

从 v1.30.0 开始,Maxwell 不再支持 JDK1.8
使用最后一个支持 JDK1.8 版本的 Maxwell v1.29.2 进行部署
wget https://github.com/zendesk/maxwell/releases/download/v1.29.2/maxwell-1.29.2.tar.gz

安装

解压即可 tar -zxvf maxwell.tar.gz

配置

vim config_1.properties

server_id=1
client_id=city_ct_63 #用于启动多个maxwell
replica_server_id=2 #用于启动多个maxwell
user=maxwell
password=123456
host=192.168.0.XX
producer=kafka
kafka.bootstrap.servers=192.168.0.XX:9092
kafka_topic=city_mysql_kafka_cdc
jdbc_options=serverTimezone=UTC
filter=exclude:*.*,include:test.mj_qyxx,include:test.mj_sbxx

vim config_2.properties

server_id=1
client_id=province_ct_63 #用于启动多个maxwell
replica_server_id=1 #用于启动多个maxwell
user=maxwell
password=123456
host=192.168.0.1x
producer=kafka
kafka.bootstrap.servers=192.168.0.xx:9092
kafka_topic=province_mysql_kafka_cdc
jdbc_options=serverTimezone=UTC
filter=exclude:*.*,include:test2.xxx

启动

bin/maxwell --config city_config.properties --daemon
bin/maxwell --config province_config.properties --daemon

验证启动进程

ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l

测试数据库全量同步

maxwell-bootstrap --database finance_result --table industry --config $MAXWELL_HOME/config.properties

其中发送过来的数据第一行及最后一行数据是标识 Maxwell 历史数据同步的,不携带任何数据。

maxwell -> kafka: 
{"database": "finance_result","table": "industry","type": "bootstrap-start","ts": 1694748250,"data": {}
}{"database": "finance_result","table": "industry","type": "bootstrap-insert","ts": 1694748250,"data": {"id": 1,"create_time": "2022-08-19 00:00:00.000000","update_time": "2022-08-19 00:00:00.000000","industry_level": 1,"industry_name": "工程建设","superior_industry_id": null}
} {"database": "finance_result","table": "industry","type": "bootstrap-insert","ts": 1694748250,"data": {"id": 2,"create_time": "2022-08-19 00:00:00.000000","update_time": "2022-08-19 00:00:00.000000","industry_level": 1,"industry_name": "轻工","superior_industry_id": null}
} {"database": "finance_result","table": "industry","type": "bootstrap-insert","ts": 1694748250,"data": {"id": 3,"create_time": "2022-08-19 00:00:00.000000","update_time": "2022-08-19 00:00:00.000000","industry_level": 2,"industry_name": "土木","superior_industry_id": 1}
}
......{"database": "finance_result","table": "industry","type": "bootstrap-complete","ts": 1694748250,"data": {}
}

测试数据库增量同步

参数说明

输出JSON字符串的格式
● data 最新的数据,修改后的数据
● old 旧数据,修改前的数据
● type 操作类型,有insert, update, delete, database-create, database-alter, database-drop, table-create, table-alter, table-drop,bootstrap-insert,int(未知类型)
● xid 事务id
● commit 同一个xid代表同一个事务,事务的最后一条语句会有commit,可以利用这个重现事务
● server_id
● thread_id
● 运行程序时添加参数–output_ddl,可以捕捉到ddl语句
● datetime列会输出为"YYYY-MM-DD hh:mm:ss",如果遇到"0000-00-00 00:00:00"会原样输出
● maxwell支持多种编码,但仅输出utf8编码
● maxwell的TIMESTAMP总是作为UTC处理,如果要调整为自己的时区,需要在后端逻辑上进行处理

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/pingmian/22919.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

2024前端面试准备2-JS基础知识回顾

变量类型和计算 1.值类型和引用类型的区别 常见值类型:undefined(定义undefined只能用let,不能用const)、字符串、bool、number、 Symbol; 常见引用类型: 对象, 数组、null(特殊引用类型,指针指向为空地址) 、function(特殊引用类型); 值类型的值直接存储在栈中;引用类型值存储…

如何避免Python中默认参数带来的陷阱

Python编程中,我们有时会给函数或方法提供默认参数。然而,这种做法在某些情况下可能会导致意想不到的行为,尤其是当默认参数是可变对象(例如列表、字典或类实例对象)时。本文将通过几个具体的例子来解释这个问题&#…

MQTT协议使用总结

MQTT是基于TCP/IP协议栈构建的异步通信消息协议,是一种轻量级的发布/订阅信息传输协议MQTT在时间和空间上,将消息发送者与接受者分离,可以在不可靠的网络环境中进行扩展。适用于设备硬件存储空间有限或网络带宽有限的场景。 物联网平台支持设…

2.1.4 采用配置类与注解方式使用MyBatis

实战概述:采用配置类与注解方式使用MyBatis 创建MyBatis配置类 在net.huawei.mybatis.config包中创建MyBatisConfig类,用于配置MyBatis核心组件,包括数据源、事务工厂和环境设置。 配置数据源和事务 使用PooledDataSource配置MySQL数据库连接…

苍穹外卖中的Druid问题

一、jar包启动解决报错 discard long time none recevied connection 升级druid版本到1.2.9 更新maven包

kafka-消费者服务搭建配置简单消费(SpringBoot整合Kafka)

文章目录 1、使用efak 创建 主题 my_topic1 并建立6个分区并给每个分区建立3个副本2、创建生产者发送消息3、application.yml配置4、创建消费者监听器5、创建SpringBoot启动类6、屏蔽 kafka debug 日志 logback.xml7、引入spring-kafka依赖 1、使用efak 创建 主题 my_topic1 并…

DP动态规划(上)

文章目录 动态规划基本概念斐波那契数列问题C 实现Python 实现Java 实现 迷你结C、Python和Java在实现动态规划时有哪些性能差异?迷你结哪种语言在动态规划中更适合大规模数据处理?迷你结C有哪些知名的库适用于动态规划和大数据处理?动态规划辅助库大数据处理库 迷…

【网络协议 | HTTP】HTTP总结与全梳理(一) —— HTTP协议超详细教程

🔥博客简介:开了几个专栏,针对 Linux 和 rtos 系统,嵌入式开发和音视频开发,结合多年工作经验,跟大家分享交流嵌入式软硬件技术、音视频技术的干货。   ✍️系列专栏:C/C、Linux、rtos、嵌入式…

二叉树练习题(2024/6/5)

1翻转二叉树 给你一棵二叉树的根节点 root ,翻转这棵二叉树,并返回其根节点。 示例 1: 输入:root [4,2,7,1,3,6,9] 输出:[4,7,2,9,6,3,1]示例 2: 输入:root [2,1,3] 输出:[2,3,1]…

dirfuzz-web敏感目录文件扫描工具

dirfuzz介绍 dirfuzz是一款基于Python3的敏感目录文件扫描工具,借鉴了dirsearch的思路,扬长避短。在根据自身实战经验的基础上而编写的一款工具,经过断断续续几个月的测试、修改和完善。 项目地址:https://github.com/ssrc-c/di…

运维开发介绍

目录 1.什么是运维开发 2.作用 3.优点 4.缺点 5.应用场景 5.1.十个应用场景 5.2.网站和Web应用程序 6.案例 7.小结 1.什么是运维开发 运维开发(DevOps)是一种结合软件开发(Development)与信息技术运维(Opera…

什么是回调函数?callback()

首先要知道一个点就是 在js中&#xff0c;函数是可以作为函数的参数传递的 所以其实回调函数 就是这个传进去的参数 其实回调函数的本质样子和普通函数是一样的 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8">…

embedding层的理解

一文读懂Embedding的概念&#xff0c;以及它和深度学习的关系 - 知乎 (zhihu.com) 感觉这篇知乎真的大道至简。个人感觉embedding层和普通的线性层没有什么区别~就是为了降维和升维用的。也就是向量的维度变化&#xff01;

基于SOA海鸥优化算法的三维曲面最高点搜索matlab仿真

目录 1.程序功能描述 2.测试软件版本以及运行结果展示 3.核心程序 4.本算法原理 5.完整程序 1.程序功能描述 基于SOA海鸥优化算法的三维曲面最高点搜索matlab仿真&#xff0c;输出收敛曲线以及三维曲面最高点搜索结果。 2.测试软件版本以及运行结果展示 MATLAB2022A版本…

Wireshark抓包工具使用 项目实战

Wireshark 是一个开源的网络协议分析器&#xff0c;它可以让你捕获和分析网络数据包&#xff0c;帮助你诊断网络问题、监控网络流量、分析协议和进行安全审计。以下是一些基本的 Wireshark 用法&#xff1a; 捕获数据包&#xff1a; 打开 Wireshark&#xff0c;选择要捕获数据包…

国外创意二维码营销案例:巴西宠物食品品牌户外活动“救救宠物爪子吧”

2024年5月份&#xff0c;巴西宠物食品品牌Purina Brasil 与广告公司Publicis Brasil合作&#xff0c;推出了一次特别的户外营销活动——Salve as Patinhas(Save the Paws)&#xff08;救救宠物爪子吧&#xff09;&#xff0c;非常有意思&#xff01; 随着全球气候变暖&#xf…

C++240605

设计一个 Per类&#xff0c;类中包含**私有**成员:姓名、年龄、**指针成员**身高、体重&#xff0c; 再设计一个Stu类&#xff0c;类中包含**私有**成员:成绩、 Per类对象p1&#xff0c; 设计这 两个类 的 **构造函数、析构函数**。 #include <iostream>using namespace…

PPT文件损坏且无法读取怎样修复?文档损坏修复方法推荐

PPT文件已经成为工作汇报、商务演示、学术交流以及教学培训中最常用到的文件&#xff0c;随着文件数量的增多和存储设备的频繁使用&#xff0c;我们有时会遇到PPT文件损坏无法打开的情况&#xff0c;这无疑给工作和学习带来了极大的困扰。 PPT文件损坏的原因可能多种多样&#…

VS2022,DLL1调用lib,lib调用DLL2

DLL1调用lib&#xff0c;lib调用DLL2 问题1&#xff1a;为什么在dll1中需要引入dll2的.lib文件 当你有一个工程&#xff08;dll1&#xff09;调用静态库&#xff08;lib&#xff09;&#xff0c;而静态库&#xff08;lib&#xff09;又调用另一个DLL&#xff08;dll2&#xf…

甜蜜约会网页制作html

创建一个简单的“甜蜜约会”主题网页&#xff0c;你需要编写HTML代码来定义网页的结构和内容。下面是一个基本的HTML页面示例&#xff0c;它展示了一个简单的“甜蜜约会”网页布局&#xff1a; <!DOCTYPE html> <html lang"zh-CN"> <head><met…