Java实现MySQL数据实时同步至Elasticsearch的方法详解

# 修改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客户端
        RestHighLevelClient esClient = ESClientFactory.createClient();
 
        // 配置Binlog监听
        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(); // 启动监听
    }
 
    private static void handleWriteEvent(WriteRowsEventData eventData, RestHighLevelClient esClient) {
        eventData.getRows().forEach(row -> {
            // 假设表结构为：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);
        });
    }
 
    // 更新和删除处理类似，代码略（完整源码见文末链接）
}