Canal 是一個用于實時同步 MySQL 數據到其他系統的工具，例如 Elasticsearch (ES)。以下是使用 Canal 將 MySQL 數據同步到 ES 的基本步驟：
確保你已經安裝并配置了 MySQL 服務器。
確保你已經安裝并配置了 Elasticsearch 服務器。
Kibana 是一個用于與 Elasticsearch 交互的 Web 界面。雖然這不是必需的，但它對于查看和管理 ES 中的數據非常有用。
a. 下載并解壓縮 Canal。
b. 修改 conf/canal.properties 文件，設置 canal.ip 和 canal.port 為你的服務器 IP 和端口。
c. 修改 conf/example/instance.properties 文件，設置以下參數：
canal.instance.master.address=<your_mysql_host>:<your_mysql_port>
canal.instance.dbUsername=<your_mysql_username>
canal.instance.dbPassword=<your_mysql_password>
canal.instance.connectionCharset=UTF-8
canal.instance.tsdb.enable=true
a. 創建一個新的 Java 項目，并添加以下依賴項：
<!-- https://mvnrepository.com/artifact/com.alibaba.otter/canal.client --><dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.5</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.elasticsearch.client/elasticsearch-rest-high-level-client --><dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.10.2</version>
</dependency>
b. 創建一個類，實現 com.alibaba.otter.canal.client.CanalConnector 接口，并在其中實現數據同步邏輯。以下是一個簡單的示例：
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestHighLevelClient;
public class MySqlToElasticsearchSync {
public static void main(String[] args) {
// 創建一個連接器
String canalHost = "localhost";
int canalPort = 11111;
String destination = "example";
String username = "";
String password = "";
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalHost, canalPort), destination, username, password);
// 連接到 Elasticsearch
RestHighLevelClient esClient = new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200, "http")));
// 訂閱數據庫表
connector.subscribe(".*\\..*");
while (true) {
// 獲取數據庫變更事件
Message message = connector.get(1024);
List<Entry> entries = message.getEntries();
// 處理每個事件
for (Entry entry : entries) {
if (entry.getEntryType() == EntryType.ROWDATA) {
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
EventType eventType = rowChange.getEventType();
// 根據事件類型進行相應的操作
switch (eventType) {
case INSERT:
case UPDATE:
// 將數據同步到 Elasticsearch
BulkRequest bulkRequest = new BulkRequest();
for (RowData rowData : rowChange.getRowDatasList()) {
Map<String, Object> dataMap = new HashMap<>();
for (Column column : rowData.getAfterColumnsList()) {
dataMap.put(column.getName(), column.getValue());
}
IndexRequest indexRequest = new IndexRequest("your_index_name").source(dataMap);
bulkRequest.add(indexRequest);
}
esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
break;
case DELETE:
// 從 Elasticsearch 中刪除數據
// ...
break;
default:
break;
}
}
}
// 確認已處理的事件
connector.ack(message.getId());
}
}
}
運行上面的 Java 程序，它將開始監聽 MySQL 數據庫的變更事件，并將數據同步到 Elasticsearch。
注意：這只是一個簡單的示例，實際應用中可能需要根據具體需求進行調整。例如，你可能需要處理更復雜的數據結構、關聯關系或者特定的業務邏輯。
