[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7
gtid_mode = ON
enforce_gtid_consistency = ON

# 重启MySQL
$ systemctl restart mysqld

# 创建Debezium用户
$ mysql -uroot -p

mysql> CREATE USER ‘debezium’@’%’ IDENTIFIED BY ‘debezium123’;
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘debezium’@’%’;
mysql> FLUSH PRIVILEGES;

# 验证权限
mysql> SHOW GRANTS FOR ‘debezium’@’%’;

输出示例如下：
+————————————————————————–+
| Grants for debezium@% |
+————————————————————————–+
| GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘debezium’@’%’ |
+————————————————————————–+

步骤2：创建MySQL连接器

{
“name”: “fgedu-mysql-connector”,
“config”: {
“connector.class”: “io.debezium.connector.mysql.MySqlConnector”,
“database.hostname”: “192.168.1.51”,
“database.port”: “3306”,
“database.user”: “debezium”,
“database.password”: “debezium123”,
“database.server.id”: “184054”,
“topic.prefix”: “fgedu_mysql”,
“database.include.list”: “fgedu_db”,
“table.include.list”: “fgedu_db.orders,fgedu_db.users”,
“database.history.kafka.bootstrap.servers”: “192.168.1.51:9092”,
“database.history.kafka.topic”: “schema-changes.fgedu”,
“include.schema.changes”: “true”,
“snapshot.mode”: “initial”,
“snapshot.locking.mode”: “minimal”,
“binlog.snapshot.locking.mode”: “minimal”,
“decimal.handling.mode”: “double”,
“time.precision.mode”: “connect”,
“tombstones.on.delete”: “true”,
“message.key.columns”: “fgedu_db.orders:id,fgedu_db.users:id”
}
}

# 注册连接器
$ curl -X POST -H “Content-Type: application/json” –data @/fgeudb/debezium/mysql-connector.json http://192.168.1.51:8083/connectors

输出示例如下：
{
“name”: “fgedu-mysql-connector”,
“config”: {
“connector.class”: “io.debezium.connector.mysql.MySqlConnector”,
“database.hostname”: “192.168.1.51”,
…
},
“tasks”: [],
“type”: “source”
}

步骤3：验证连接器状态

输出示例如下：
{
“name”: “fgedu-mysql-connector”,
“connector”: {
“state”: “RUNNING”,
“worker_id”: “192.168.1.51:8083”
},
“tasks”: [
{
“id”: 0,
“state”: “RUNNING”,
“worker_id”: “192.168.1.51:8083”
}
],
“type”: “source”
}

# 查看生成的Topic
$ /fgeudb/kafka/bin/kafka-topics.sh –bootstrap-server 192.168.1.51:9092 –list | grep fgedu

输出示例如下：
fgedu_mysql
fgedu_mysql.fgedu_db.orders
fgedu_mysql.fgedu_db.users

步骤4：消费变更事件

输出示例如下：
{
“schema”: {…},
“payload”: {
“before”: null,
“after”: {
“id”: 1,
“order_no”: “ORD20260404001”,
“user_id”: 100,
“amount”: 999.99,
“status”: “pending”,
“create_time”: “2026-04-04T10:00:00Z”
},
“source”: {
“version”: “2.7.0.Final”,
“connector”: “mysql”,
“name”: “fgedu_mysql”,
“ts_ms”: 1712217600000,
“db”: “fgedu_db”,
“table”: “orders”,
“server_id”: 1,
“gtid”: “3a4e7b00-1234-5678-90ab-cdef12345678:100”,
“file”: “mysql-bin.000001”,
“pos”: 1234,
“row”: 0
},
“op”: “c”,
“ts_ms”: 1712217600123
}
}

wal_level = logical
max_wal_senders = 4
max_replication_slots = 4
wal_sender_timeout = 60s

# 编辑pg_hba.conf
$ vi /fgeudb/postgresql/data/pg_hba.conf

host replication debezium 192.168.1.0/24 md5
host all debezium 192.168.1.0/24 md5

# 重启PostgreSQL
$ systemctl restart postgresql

# 创建Debezium用户和插件
$ psql -U postgres

postgres=# CREATE USER debezium WITH REPLICATION PASSWORD ‘debezium123’;
postgres=# CREATE DATABASE fgedu_db;
postgres=# \c fgedu_db
fgedu_db=# CREATE SCHEMA fgedu;
fgedu_db=# GRANT ALL ON SCHEMA fgedu TO debezium;
fgedu_db=# ALTER USER debezium SET search_path TO fgedu, public;

# 安装pgoutput插件（PostgreSQL 10+内置）
fgedu_db=# SELECT * FROM pg_create_logical_replication_slot(‘debezium_slot’, ‘pgoutput’);

输出示例如下：
slot_name | lsn
—————-+———–
debezium_slot | 0/16B99D0

步骤2：创建PostgreSQL连接器

{
“name”: “fgedu-postgres-connector”,
“config”: {
“connector.class”: “io.debezium.connector.postgresql.PostgresConnector”,
“database.hostname”: “192.168.1.51”,
“database.port”: “5432”,
“database.user”: “debezium”,
“database.password”: “debezium123”,
“database.dbname”: “fgedu_db”,
“topic.prefix”: “fgedu_pg”,
“schema.include.list”: “fgedu”,
“table.include.list”: “fgedu.orders,fgedu.users”,
“plugin.name”: “pgoutput”,
“slot.name”: “debezium_slot”,
“publication.name”: “dbz_publication”,
“publication.autocreate.mode”: “filtered”,
“snapshot.mode”: “initial”,
“decimal.handling.mode”: “double”,
“hstore.handling.mode”: “json”,
“interval.handling.mode”: “numeric”,
“database.tcpkeepalive”: “true”
}
}

# 注册连接器
$ curl -X POST -H “Content-Type: application/json” –data @/fgeudb/debezium/postgres-connector.json http://192.168.1.51:8083/connectors

输出示例如下：
{
“name”: “fgedu-postgres-connector”,
“config”: {…},
“tasks”: [],
“type”: “source”
}

# 开启归档模式
SQL> SHUTDOWN IMMEDIATE;
SQL> STARTUP MOUNT;
SQL> ALTER DATABASE ARCHIVELOG;
SQL> ALTER DATABASE OPEN;

# 开启补充日志
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
SQL> ALTER SYSTEM SWITCH LOGFILE;

# 创建Debezium用户
SQL> CREATE USER debezium IDENTIFIED BY debezium123;
SQL> GRANT CONNECT, RESOURCE TO debezium;
SQL> GRANT SELECT ANY TABLE TO debezium;
SQL> GRANT SELECT ANY TRANSACTION TO debezium;
SQL> GRANT EXECUTE ON SYS.DBMS_LOGMNR TO debezium;
SQL> GRANT SELECT ON V_$LOGMNR_CONTENTS TO debezium;
SQL> GRANT SELECT ON V_$DATABASE TO debezium;
SQL> GRANT SELECT ON V_$LOG TO debezium;
SQL> GRANT SELECT ON V_$LOGFILE TO debezium;
SQL> GRANT SELECT ON V_$ARCHIVED_LOG TO debezium;

# 验证配置
SQL> SELECT NAME, LOG_MODE FROM V$DATABASE;

输出示例如下：
NAME LOG_MODE
——— ————
ARCHIVELOG

步骤2：创建Oracle连接器

{
“name”: “fgedu-oracle-connector”,
“config”: {
“connector.class”: “io.debezium.connector.oracle.OracleConnector”,
“database.hostname”: “192.168.1.51”,
“database.port”: “1521”,
“database.user”: “debezium”,
“database.password”: “debezium123”,
“database.dbname”: “”,
“topic.prefix”: “fgedu_oracle”,
“schema.include.list”: “”,
“table.include.list”: “.ORDERS,.USERS”,
“database.connection.adapter”: “logminer”,
“log.mining.strategy”: “online_catalog”,
“log.mining.archive.log.hours”: 24,
“log.mining.archive.log.only.mode”: “false”,
“snapshot.mode”: “initial”,
“decimal.handling.mode”: “double”,
“lob.enabled”: “false”,
“unified.value.schemas”: “true”
}
}

# 注册连接器
$ curl -X POST -H “Content-Type: application/json” –data @/fgeudb/debezium/oracle-connector.json http://192.168.1.51:8083/connectors