MySQL数据实时同步至Elasticsearch的高效方案：Java实现+源码解析，一文搞定！

引言：为什么需要实时同步？

MySQL擅长事务处理，而Elasticsearch（ES）则专注于搜索与分析。将MySQL数据实时同步到ES，可以充分发挥两者的优势，例如：

构建高性能搜索服务
实时数据分析与大屏展示
提升复杂查询效率

传统方案（如定时全量同步）存在延迟高、资源浪费等问题。本文将基于MySQL Binlog监听实现毫秒级实时同步，并提供完整Java代码及深度源码解析。

一、技术选型与核心原理

1.1 核心组件

MySQL Binlog：MySQL的二进制日志，记录所有数据变更事件（增删改）。
Canal/OpenReplicator：解析Binlog的工具（本文使用轻量级mysql-binlog-connector-java）。
Elasticsearch High Level REST Client：ES官方Java客户端，用于数据写入。

1.2 架构流程图

MySQL Server → Binlog → Java监听程序 → 数据转换 → Elasticsearch

二、环境准备与配置

2.1 MySQL开启Binlog

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

2.2 创建ES索引

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

三、Java代码实现

3.1 Maven依赖

<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>

3.2 核心代码（Binlog监听与同步）

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_timeString 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);});}// 更新和删除处理类似，代码略（完整源码见文末链接）
}