docker 安装mysql 参考： docker 安装mysql 8.x_大大蚊子的博客-CSDN博客

docker 安装kafka 参考：  docker 安装 kafka单节点_docker kafka 单节点_大大蚊子的博客-CSDN博客

docker 安装debezium参考：debezium docker 容器创建_大大蚊子的博客-CSDN博客

环境描述：

已安装mysql数据库

已安装kafka

已安装debezium

检查mysql是否满足debezium接入

1、log_bin 开启

show variables like 'log_%';

 2、binlog_format 为ROW

show variables like '%format%';

 如不满足，修改mysql 配置文件

查找my.cnf 文件

mysqld --verbose --help | grep -A 1 'options'

修改以下配置并重启

log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 10

service mysqld restart

创建CDC专属用户（可选，测试可以直接使用root用户）

 5.7

grant select,reload,show databases,replication slave,replication client on *.* to 'dbz_user'@'%' identified by 'dbz_user';
flush privileges;

8.X

create user 'dbz_user'@'%' identified by 'dbz_user';
grant select,reload,show databases,replication slave,replication client on *.* to 'dbz_user'@'%' with grant option;
flush privileges;

开启采集作业

 postman

接口一： POST 创建作业   查询serverid    SELECT@@server_id  

http://192.168.1.5:8083/connectors{"name": "mysql_test_01", "config": {"connector.class": "io.debezium.connector.mysql.MySqlConnector", "database.hostname": "192.168.1.106", "database.port": "3306", "database.user": "root", "database.password": "123456", "database.server.id": "1", "database.server.name": "mysql_test", "database.include.list": "dbz_mysql", "database.history.kafka.bootstrap.servers": "192.168.1.102:9092", "database.history.kafka.topic": "dbhistory.fullfillment", "include.schema.changes": "true" }
}

 接口二： get 获取作业列表   http://192.168.1.5:8083/connectors/

 mysql 操作数据

kafka 消费对应数据

/opt/bitnami/kafka/bin/kafka-console-consumer.sh --property print.key=true --property print.timestamp=ture --bootstrap-server 192.168.1.102:9092 --topic mysql_test.dbz_mysql.aa  --from-beginning{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":true,"name":"mysql_test.dbz_mysql.aa.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":true,"name":"mysql_test.dbz_mysql.aa.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"mysql_test.dbz_mysql.aa.Envelope"},"payload":{"before":{"id":2},"after":{"id":3},"source":{"version":"1.9.5.Final","connector":"mysql","name":"mysql_test","ts_ms":1678548518000,"snapshot":"false","db":"dbz_mysql","sequence":null,"table":"aa","server_id":1,"gtid":null,"file":"binlog.000006","pos":675,"row":0,"thread":11,"query":null},"op":"u","ts_ms":1678548518053,"transaction":null}}