Canal 是一個用于實時同步 MySQL 數據到其他系統的工具，例如 Elasticsearch (ES)。以下是使用 Canal 將 MySQL 數據同步到 ES 的基本步驟：

1. 安裝和配置 MySQL

確保你已經安裝并配置了 MySQL 服務器。

2. 安裝和配置 Elasticsearch

確保你已經安裝并配置了 Elasticsearch 服務器。

3. 安裝和配置 Kibana（可選）

Kibana 是一個用于與 Elasticsearch 交互的 Web 界面。雖然這不是必需的，但它對于查看和管理 ES 中的數據非常有用。

4. 安裝和配置 Canal

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

5. 創建和配置數據同步客戶端

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());
        }
    }
}

6. 運行程序

運行上面的 Java 程序，它將開始監聽 MySQL 數據庫的變更事件，并將數據同步到 Elasticsearch。

注意：這只是一個簡單的示例，實際應用中可能需要根據具體需求進行調整。例如，你可能需要處理更復雜的數據結構、關聯關系或者特定的業務邏輯。