1. Canal 與 Elasticsearch 實時同步概述Canal 是阿里巴巴開源的基于數(shù)據(jù)庫增量日志解析的組件支持 MySQL、Oracle 等數(shù)據(jù)庫。它通過解析數(shù)據(jù)庫的 binlog 日志將數(shù)據(jù)變更事件推送到消息隊列或直接處理實現(xiàn)數(shù)據(jù)的準實時同步。Elasticsearch 是一個基于 Lucene 庫的搜索引擎具有強大的全文檢索和分析能力。將 Canal 與 Elasticsearch 結(jié)合使用可以實現(xiàn)數(shù)據(jù)庫數(shù)據(jù)到 Elasticsearch 的準實時同步充分發(fā)揮 Elasticsearch 在搜索、分析和日志處理方面的優(yōu)勢。這種架構(gòu)廣泛應用于電商搜索、日志分析、監(jiān)控告警等場景。Canal 與 Elasticsearch 同步的基本架構(gòu)包括Canal Server: 監(jiān)聽并解析數(shù)據(jù)庫 binlogCanal Client: 接收變更事件并處理數(shù)據(jù)轉(zhuǎn)換: 將數(shù)據(jù)庫數(shù)據(jù)轉(zhuǎn)換為 Elasticsearch 文檔格式Elasticsearch Indexer: 將文檔寫入 Elasticsearch這種架構(gòu)的優(yōu)勢在于低延遲、高可靠性且對數(shù)據(jù)庫幾乎無侵入。2. 增量索引更新實現(xiàn)增量索引更新是同步方案的核心主要通過 Canal 監(jiān)聽數(shù)據(jù)庫變更事件并將變更應用到 Elasticsearch 索引中。實現(xiàn)增量更新的關鍵步驟2.1. 配置 Canal 監(jiān)聽數(shù)據(jù)庫首先需要在 MySQL 數(shù)據(jù)庫中開啟 binlog 功能并配置 Canal 監(jiān)聽指定數(shù)據(jù)庫。修改 my.cnf 文件添加以下配置[mysqld] server-id1 log-binmysql-bin binlog-formatROW binlog-row-imageFULL2.2. 創(chuàng)建 Canal 實例創(chuàng)建一個新的 Canal 實例指向需要同步的數(shù)據(jù)庫canal.instance.mysql.slaveId1234 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal canal.instance.dbNameyour_database canal.instance.dbEncodingUTF-82.3. 實現(xiàn)消息處理邏輯編寫 Canal Client 接收變更事件并將數(shù)據(jù)寫入 Elasticsearchpublic class ElasticsearchHandler implements EntryHandlerCanalEntry.Entry { private RestHighLevelClient esClient; public ElasticsearchHandler(RestHighLevelClient esClient) { this.esClient esClient; } Override public void handle(CanalEntry.Entry entry) throws Exception { if (entry.getEntryType() CanalEntry.EntryType.ROWDATA) { CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (rowChange.getEventType() CanalEntry.EventType.INSERT || rowChange.getEventType() CanalEntry.EventType.UPDATE) { // 處理插入和更新操作 IndexRequest request new IndexRequest(your_index) .id(rowData.getAfterColumns(0).getValue()) .source(convertToMap(rowData.getAfterColumnsList())); esClient.index(request, RequestOptions.DEFAULT); } else if (rowChange.getEventType() CanalEntry.EventType.DELETE) { // 處理刪除操作 DeleteRequest request new DeleteRequest(your_index) .id(rowData.getBeforeColumns(0).getValue()); esClient.delete(request, RequestOptions.DEFAULT); } } } } private MapString, Object convertToMap(ListCanalEntry.Column columns) { MapString, Object map new HashMap(); for (CanalEntry.Column column : columns) { if (column.getIsNull()) { map.put(column.getName(), null); } else { map.put(column.getName(), column.getValue()); } } return map; } }2.4. 處理批量提交為提高性能可以使用批量提交機制BulkRequest bulkRequest new BulkRequest(); // 添加多個索引/刪除請求到批量請求中 // ... // 執(zhí)行批量提交 BulkResponse bulkResponse esClient.bulk(bulkRequest, RequestOptions.DEFAULT); if (bulkResponse.hasFailures()) { // 處理失敗情況 }3. 數(shù)據(jù)刪除處理方案在 Canal 與 Elasticsearch 同步過程中處理刪除操作是一個關鍵點。與數(shù)據(jù)更新不同刪除操作需要特別注意數(shù)據(jù)一致性和同步延遲問題。3.1. 基于主鍵的刪除處理最簡單的刪除方式是基于主鍵進行刪除如上述代碼所示。這種方法適用于每個表都有明確主鍵的情況。3.2. 軟刪除與硬刪除根據(jù)業(yè)務需求可以選擇軟刪除或硬刪除硬刪除直接從 Elasticsearch 中刪除文檔軟刪除在文檔中標記為已刪除而不是真正刪除適合需要保留歷史數(shù)據(jù)的場景3.3. 刪除事件過濾在某些場景下可能需要過濾特定的刪除事件Override public void handle(CanalEntry.Entry entry) throws Exception { if (entry.getEntryType() CanalEntry.EntryType.ROWDATA) { CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (rowChange.getEventType() CanalEntry.EventType.DELETE) { // 檢查是否需要跳過此刪除事件 if (shouldSkipDelete(rowData)) { continue; } // 執(zhí)行刪除操作 DeleteRequest request new DeleteRequest(your_index) .id(rowData.getBeforeColumns(0).getValue()); esClient.delete(request, RequestOptions.DEFAULT); } } } } private boolean shouldSkipDelete(CanalEntry.RowData rowData) { // 根據(jù)業(yè)務邏輯判斷是否跳過刪除 // 例如特定狀態(tài)的數(shù)據(jù)不執(zhí)行刪除操作 for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { if (status.equals(column.getName()) inactive.equals(column.getValue())) { return true; } } return false; }3.4. 刪除操作的冪等性確保刪除操作的冪等性非常重要特別是在網(wǎng)絡不穩(wěn)定或重試的情況下public void safeDelete(String index, String id) { try { // 檢查文檔是否存在 GetRequest getRequest new GetRequest(index, id); boolean exists esClient.exists(getRequest, RequestOptions.DEFAULT); if (exists) { // 文檔存在則刪除 DeleteRequest deleteRequest new DeleteRequest(index, id); esClient.delete(deleteRequest, RequestOptions.DEFAULT); } // 如果文檔不存在不做任何操作 } catch (ElasticsearchException e) { // 處理異常如文檔已被其他線程刪除的情況 if (e.getDetailedMessage().contains(missing)) { // 文檔不存在無需處理 return; } throw e; } }4. 全量重建策略雖然增量同步能夠保持數(shù)據(jù)一致性但在某些情況下需要進行全量重建例如首次同步索引結(jié)構(gòu)變更數(shù)據(jù)發(fā)生嚴重不一致需要修復4.1. 全量重建方案設計全量重建的基本流程如下停止 Canal 增量同步從數(shù)據(jù)庫導出全量數(shù)據(jù)清空 Elasticsearch 索引將全量數(shù)據(jù)導入 Elasticsearch重新啟動 Canal 增量同步4.2. 實現(xiàn)全量數(shù)據(jù)導出可以使用 JDBC 直接從數(shù)據(jù)庫查詢?nèi)繑?shù)據(jù)public ListMapString, Object exportFullData(String sql, Connection connection) throws SQLException { ListMapString, Object result new ArrayList(); try (PreparedStatement stmt connection.prepareStatement(sql); ResultSet rs stmt.executeQuery()) { ResultSetMetaData metaData rs.getMetaData(); int columnCount metaData.getColumnCount(); while (rs.next()) { MapString, Object row new LinkedHashMap(); for (int i 1; i columnCount; i) { row.put(metaData.getColumnName(i), rs.getObject(i)); } result.add(row); } } return result; }4.3. 批量導入 Elasticsearch使用 Elasticsearch 的批量 API 高效導入數(shù)據(jù)public void bulkIndexToES(ListMapString, Object documents, String index) throws IOException { BulkRequest bulkRequest new BulkRequest(); for (MapString, Object doc : documents) { // 假設文檔中包含 id 字段 String id doc.get(id).toString(); // 移除 id 字段因為它在 IndexRequest 中單獨指定 doc.remove(id); IndexRequest request new IndexRequest(index).id(id).source(doc); bulkRequest.add(request); // 每 1000 條提交一次 if (bulkRequest.numberOfActions() 1000) { BulkResponse bulkResponse esClient.bulk(bulkRequest, RequestOptions.DEFAULT); if (bulkResponse.hasFailures()) { // 處理失敗 handleFailures(bulkResponse); } bulkRequest new BulkRequest(); } } // 提交剩余的請求 if (bulkRequest.numberOfActions() 0) { BulkResponse bulkResponse esClient.bulk(bulkRequest, RequestOptions.DEFAULT); if (bulkResponse.hasFailures()) { handleFailures(bulkResponse); } } } private void handleFailures(BulkResponse bulkResponse) { for (BulkItemResponse response : bulkResponse) { if (response.isFailed()) { // 記錄失敗信息 System.err.println(Failed to process document: response.getId() , Error: response.getFailure().getMessage()); } } }4.4. 全量重建與增量同步的銜接為避免數(shù)據(jù)丟失全量重建與增量同步之間需要正確銜接記錄全量數(shù)據(jù)導出的時間點 T在時間點 T 之后的所有數(shù)據(jù)庫變更需要單獨記錄全量數(shù)據(jù)導入完成后將時間點 T 之后的增量變更同步到 Elasticsearch這可以通過記錄 binlog 位置來實現(xiàn)public class BinlogPosition { private String logFileName; private long logFileOffset; // 獲取方法 public String getLogFileName() { return logFileName; } public long getLogFileOffset() { return logFileOffset; } // 設置方法 public void setLogFileName(String logFileName) { this.logFileName logFileName; } public void setLogFileOffset(long logFileOffset) { this.logFileOffset logFileOffset; } } // 在全量導出開始前記錄位置 public BinlogPosition getCurrentBinlogPosition(Connection mysqlConn) throws SQLException { BinlogPosition position new BinlogPosition(); try (Statement stmt mysqlConn.createStatement(); ResultSet rs stmt.executeQuery(SHOW MASTER STATUS)) { if (rs.next()) { position.setLogFileName(rs.getString(File)); position.setLogFileOffset(rs.getLong(Position)); } } return position; } // 在全量導入完成后從此位置繼續(xù)同步 public void resumeIncrementalSync(BinlogPosition position) { // 配置 Canal 從指定位置開始監(jiān)聽 canalConfig.setMasterId(position.getLogFileName()); canalConfig.setSlaveId(position.getLogFileOffset()); // 啟動 Canal 客戶端 startCanalClient(); }5. 實踐案例與注意事項5.1. 完整的最小示例下面是一個完整的 Canal 與 Elasticsearch 同步的最小示例public class CanalElasticsearchSync { private RestHighLevelClient esClient; private CanalClient canalClient; public void init() { // 初始化 Elasticsearch 客戶端 esClient new RestHighLevelClient( RestClient.builder(new HttpHost(localhost, 9200, http))); // 初始化 Canal 客戶端 canalClient new CanalConnector(localhost, 11111, canal, canal, example); canalClient.connect(); canalClient.subscribe(); canalClient.rollback(); } public void startSync() { try { while (true) { Message message canalClient.getWithoutAck(100); long batchId message.getId(); if (message.getEntries() ! null !message.getEntries().isEmpty()) { for (CanalEntry.Entry entry : message.getEntries()) { handleEntry(entry); } } canalClient.ack(batchId); } } catch (Exception e) { e.printStackTrace(); } finally { canalClient.disconnect(); try { esClient.close(); } catch (IOException e) { e.printStackTrace(); } } } private void handleEntry(CanalEntry.Entry entry) throws Exception { if (entry.getEntryType() CanalEntry.EntryType.ROWDATA) { CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (rowChange.getEventType() CanalEntry.EventType.INSERT || rowChange.getEventType() CanalEntry.EventType.UPDATE) { // 處理插入和更新 IndexRequest request new IndexRequest(your_index) .id(rowData.getAfterColumns(0).getValue()) .source(convertColumnsToMap(rowData.getAfterColumnsList())); esClient.index(request, RequestOptions.DEFAULT); } else if (rowChange.getEventType() CanalEntry.EventType.DELETE) { // 處理刪除 DeleteRequest request new DeleteRequest(your_index) .id(rowData.getBeforeColumns(0).getValue()); esClient.delete(request, RequestOptions.DEFAULT); } } } } private MapString, Object convertColumnsToMap(ListCanalEntry.Column columns) { MapString, Object map new HashMap(); for (CanalEntry.Column column : columns) { if (column.getIsNull()) { map.put(column.getName(), null); } else { map.put(column.getName(), column.getValue()); } } return map; } public static void main(String[] args) { CanalElasticsearchSync sync new CanalElasticsearchSync(); sync.init(); sync.startSync(); } }5.2. 注意事項性能監(jiān)控監(jiān)控 Canal 和 Elasticsearch 的性能指標及時發(fā)現(xiàn)并處理性能瓶頸錯誤處理建立完善的錯誤處理機制特別是網(wǎng)絡中斷、數(shù)據(jù)格式錯誤等情況數(shù)據(jù)一致性定期檢查 Canal 和 Elasticsearch 之間的數(shù)據(jù)一致性特別是關鍵業(yè)務數(shù)據(jù)備份策略制定并執(zhí)行數(shù)據(jù)備份策略防止數(shù)據(jù)丟失版本兼容性確保 Canal 和 Elasticsearch 版本兼容避免因版本不匹配導致的問題資源管理合理配置內(nèi)存和線程資源避免資源耗盡導致系統(tǒng)崩潰5.3. 同步策略對比| 同步策略 | 優(yōu)點 | 缺點 | 適用場景 ||---------|------|------|---------|| 實時同步 | 低延遲數(shù)據(jù)最新 | 對數(shù)據(jù)庫壓力大資源消耗高 | 對實時性要求高的場景 || 批量同步 | 資源消耗小系統(tǒng)穩(wěn)定性高 | 同步延遲高 | 對實時性要求不高的場景 || 混合策略 | 平衡實時性與資源消耗 | 實現(xiàn)復雜度高 | 大規(guī)模數(shù)據(jù)同步場景 |下面是 Canal 與 Elasticsearch 實時同步的流程圖Binlog變更數(shù)據(jù)處理數(shù)據(jù)監(jiān)控監(jiān)控監(jiān)控MySQL數(shù)據(jù)庫Canal服務器消息隊列/處理程序Elasticsearch監(jiān)控工具