MySQL數據變更Kafka實時捕獲

要實現MySQL數據變更實時捕獲并發送到Kafka，你可以使用一些開源工具，如Debezium、Canal等。這里以Debezium為例，介紹如何實現這個功能。

1. 安裝Debezium

首先，你需要在你的MySQL服務器和Kafka服務器上安裝Debezium。具體安裝方法可以參考官方文檔：https://debezium.io/quickstart/

2. 配置Debezium

在MySQL服務器上，創建一個名為my_database的數據庫，并創建一個名為my_table的表：

CREATE DATABASE my_database;
USE my_database;

CREATE TABLE my_table (
  id INT AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(255),
  age INT
);

接下來，在Debezium的配置文件（如connect-mysql.properties）中，配置MySQL連接信息和Kafka主題信息：

connector.type=mysql
connector.url=jdbc:mysql://localhost:3306/my_database?serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true
connector.table=my_table
connector.topic=my_table_changes
connector.mode=schema-only

這里，connector.type指定了連接類型為MySQL，connector.url指定了MySQL服務器的連接信息，connector.table指定了要監聽的表名，connector.topic指定了Kafka主題名，connector.mode指定了監聽模式為schema-only，表示只監聽表結構變化。

3. 啟動Debezium

使用以下命令啟動Debezium：

bin/connect connect-mysql.properties

此時，Debezium將開始監聽my_table的表結構變化。

4. 配置Kafka消費者

創建一個名為my_consumer.properties的Kafka消費者配置文件：

bootstrap.servers=localhost:9092
group.id=my_consumer_group
auto.offset.reset=earliest
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

這里，bootstrap.servers指定了Kafka服務器地址，group.id指定了消費者組ID，auto.offset.reset指定了消費者啟動時的初始偏移量，key.deserializer和value.deserializer指定了鍵值對的反序列化方式。

5. 啟動Kafka消費者

使用以下命令啟動Kafka消費者：

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my_consumer_group --from-beginning my_consumer.properties

此時，Kafka消費者將開始消費my_table_changes主題的消息，包括表結構變化和記錄變更。

通過以上步驟，你可以實現MySQL數據變更實時捕獲并發送到Kafka的功能。