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

MySQL Binlog 监听方案

news 2025/1/7 12:37:34

如果 EmbeddedEngine 类在 debezium-connector-mysql 中不可用，原因是 Debezium 的新版本移除了 EmbeddedEngine。这是因为 Debezium 的架构变更，它现在鼓励使用 Kafka Connect 或 Debezium Server 来处理数据变更事件。

下面是几种替代方法来实现 MySQL Binlog 监听，你可以根据项目需求选择合适的方案。

✅ 替代方案 1：使用 Debezium Server 监听 Binlog

Debezium Server 是一个独立的服务，可以将 MySQL Binlog 的变更事件发送到目标系统，如 Kafka、Pulsar、Google Pub/Sub 等。你可以将它作为一个独立服务运行，并将结果集成到 Spring Boot 中。

🔧 步骤：

1️⃣ 下载 Debezium Server

从官方 GitHub 下载最新的 Debezium Server：

👉 Debezium Server Releases

2️⃣ 配置 application.properties

在 Debezium Server 的配置文件中配置 MySQL 连接 和 目标系统（如 Kafka）。

debezium.sink.type=kafka
debezium.sink.kafka.bootstrap.servers=localhost:9092debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.database.hostname=localhost
debezium.source.database.port=3306
debezium.source.database.user=root
debezium.source.database.password=your_password
debezium.source.database.server.id=85744
debezium.source.database.include.list=your_database
debezium.source.database.history.kafka.bootstrap.servers=localhost:9092
debezium.source.database.history.kafka.topic=schema-changes.your_database

3️⃣ 运行 Debezium Server

java -jar debezium-server-<version>.jar

4️⃣ 在 Spring Boot 中消费 Kafka 消息

你可以在 Spring Boot 中使用 Kafka 监听来自 Debezium 的事件。

添加依赖：

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

创建 Kafka 消费者：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;@Service
public class KafkaBinlogConsumer {@KafkaListener(topics = "your_database.your_table", groupId = "binlog-group")public void listen(String message) {System.out.println("Received: " + message);// 解析并处理 Binlog 数据}
}

✅ 替代方案 2：使用 Kafka Connect

如果你已经使用 Kafka，可以通过 Kafka Connect 部署 Debezium MySQL Connector 来监听 MySQL Binlog。

🔧 配置步骤：

1️⃣ 启动 Kafka Connect

确保 Kafka Connect 已启动：

./bin/connect-distributed.sh config/connect-distributed.properties

2️⃣ 配置 MySQL Connector

创建一个 Debezium MySQL Connector 的 JSON 文件：

{"name": "mysql-connector","config": {"connector.class": "io.debezium.connector.mysql.MySqlConnector","database.hostname": "localhost","database.port": "3306","database.user": "root","database.password": "your_password","database.server.id": "1","database.include.list": "your_database","database.history.kafka.bootstrap.servers": "localhost:9092","database.history.kafka.topic": "schema-changes.your_database"}
}

将该配置文件 POST 到 Kafka Connect 的 REST API：

curl -X POST -H "Content-Type: application/json" --data @mysql-connector.json http://localhost:8083/connectors

✅ 替代方案 3：使用 Maxwell for Binlog 监听

Maxwell 是一个轻量级的工具，用于将 MySQL Binlog 转换成 JSON 格式 并发送到 Kafka、Pulsar、Redis、Elasticsearch 等系统。

👉 Maxwell GitHub

🔧 步骤：

1️⃣ 下载并配置 Maxwell

wget https://github.com/zendesk/maxwell/releases/download/v1.40.0/maxwell-1.40.0.tar.gz
tar -xvf maxwell-1.40.0.tar.gz

2️⃣ 运行 Maxwell

bin/maxwell --user='root' --password='your_password' --host='localhost' --producer=kafka

✅ 替代方案 4：使用 Canal for Binlog 监听

Alibaba Canal 是一个用于监听 MySQL Binlog 的工具，支持将变更数据发送到 Kafka、RabbitMQ 等消息队列中。

👉 Canal GitHub

✅ 总结对比

工具	使用场景	优势	备注
Debezium Server	不依赖 Kafka	直接将数据推送到目标系统	官方推荐的架构变更方式
Kafka Connect	使用 Kafka 的场景	生态系统丰富，支持多种数据源	适合大规模数据同步
Maxwell	轻量级 JSON Binlog 监听器	简单易用，配置方便	适合小型项目
Canal	高性能、支持多种目标系统	支持多种 MQ，性能较高	需要更多配置