Apache Kafka 是一款分布式消息系統，可以實現快速、可靠地傳輸消息，并被廣泛用于數據處理、微服務等場景。其中一個重要的應用場景就是將 MySQL 數據同步到 Elasticsearch。

具體實現方法如下：

1. 創建Kafka Topic
// 以 MySQL 數據表 t_user 為例，創建一個名為 t_user 的 Kafka Topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic t_user
2. 使用Debezium實現MySQL數據變更的監控
// Debezium 是一個基于Kafka的分布式數據變更事件捕獲系統，通過CDC技術實現實時監聽 MySQL 數據庫的變更并將其轉換為Kafka消息。
// 下載Debezium Connector for MySQL，并將其解壓到Kafka的插件目錄中
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.5.0.Final/debezium-connector-mysql-1.5.0.Final-plugin.tar.gz
tar xvf debezium-connector-mysql-1.5.0.Final-plugin.tar.gz -C /path/to/kafka/plugins/
// 修改Kafka配置文件 config/connect-distributed.properties，開啟Debezium插件
plugin.path=/path/to/kafka/plugins
// 在Kafka集群中啟動兩個Kafka Connect實例
bin/connect-distributed.sh config/connect-distributed.properties
// 創建一個Debezium Connector配置文件 t_user_connector.json，用于監聽 t_user 表的變更，并將其轉換為Kafka消息
{
"name": "t_user_connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "localhost",
"database.port": "3306",
"database.user": "root",
"database.password": "root",
"database.server.id": "1",
"database.server.name": "t_user_mysql",
"database.history.kafka.bootstrap.servers": "localhost:9092",
"database.history.kafka.topic": "dbhistory.t_user",
"table.include.list": "test.t_user",
"database.history.skip.unparseable.ddl": true,
"include.schema.changes": true
}
}
// 使用Kafka Connect提交該配置文件，啟動 t_user 表的監聽
bin/connect-standalone.sh config/connect-distributed.properties t_user_connector.json
3. 使用Logstash將Kafka消息寫入Elasticsearch
// Logstash 是一款用于數據處理的開源工具，支持從 Kafka 中讀取消息并將其寫入 Elasticsearch。
// 下載Logstash，并創建一個配置文件，用于從 Kafka 中讀取 t_user 表的消息并將其寫入 Elasticsearch。
input {
kafka {
bootstrap_servers =>"localhost:9092"
topics =>["t_user"]
}
}
output {
elasticsearch {
hosts =>["localhost:9200"]
index =>"t_user-%{+YYYY.MM.dd}"
}
}
// 使用Logstash提交該配置文件，啟動消息的讀取和寫入
bin/logstash -f t_user.conf

至此，MySQL 數據就被實時同步到 Elasticsearch 中了。Kafka 提供了高吞吐量、低延時的消息傳輸能力，使得數據同步變得更加穩定、可靠。