# 修改my.cnf（Linux）或my.ini（Windows）
[mysqld]
server_id=1
log_bin=mysql-bin
binlog_format=ROW  # 必須為ROW模式

PUT /user
{
  "mappings": {
    "properties": {
      "id": {"type": "integer"},
      "name": {"type": "text"},
      "email": {"type": "keyword"},
      "create_time": {"type": "date"}
    }
  }
}

<dependency>
    <groupId>com.github.shyiko</groupId>
    <artifactId>mysql-binlog-connector-java</artifactId>
    <version>0.25.4</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.17.3</version>
</dependency>

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
 
public class MySQL2ESSyncer {
 
    private static final String ES_INDEX = "user";
 
    public static void main(String[] args) throws Exception {
        // 初始化ES客戶(hù)端
        RestHighLevelClient esClient = ESClientFactory.createClient();
 
        // 配置Binlog監(jiān)聽(tīng)
        BinaryLogClient client = new BinaryLogClient("localhost", 3306, "root", "password");
        client.setServerId(1001); // 唯一ID，避免沖突
 
        client.registerEventListener(event -> {
            EventData data = event.getData();
            if (data instanceof WriteRowsEventData) {
                // 處理插入事件
                handleWriteEvent((WriteRowsEventData) data, esClient);
            } else if (data instanceof UpdateRowsEventData) {
                // 處理更新事件
                handleUpdateEvent((UpdateRowsEventData) data, esClient);
            } else if (data instanceof DeleteRowsEventData) {
                // 處理刪除事件
                handleDeleteEvent((DeleteRowsEventData) data, esClient);
            }
        });
 
        client.connect(); // 啟動(dòng)監(jiān)聽(tīng)
    }
 
    private static void handleWriteEvent(WriteRowsEventData eventData, RestHighLevelClient esClient) {
        eventData.getRows().forEach(row -> {
            // 假設(shè)表結(jié)構(gòu)為：id, name, email, create_time
            String json = String.format(
                "{\"id\":%d,\"name\":\"%s\",\"email\":\"%s\",\"create_time\":\"%s\"}",
                row[0], row[1], row[2], row[3]
            );
            IndexRequest request = new IndexRequest(ES_INDEX)
                .id(row[0].toString())
                .source(json, XContentType.JSON);
            esClient.index(request, RequestOptions.DEFAULT);
        });
    }
 
    // 更新和刪除處理類(lèi)似，代碼略（完整源碼見(jiàn)文末鏈接）
}