相關鏈接：

ElasticSearch官網地址：https://www.elastic.co/cn/

Lucene官網地址：https://lucene.apache.org/

IK分詞器官網地址：https://github.com/infinilabs/analysis-ik

Canal官網地址：https://github.com/alibaba/canal

一、為什么選用ElasticSearch

ElasticSearch是一款非常強大的開源的分布式搜索引擎，具備從海量數據中快速找到需要內容的功能，可以用來實現搜索、日志統計、分析、系統監控等功能。

在實際項目開發中，我們經常將Mysql作為業務數據庫，ES作為查詢數據庫，一是可以用來實現讀寫分離，使項目的架構有更好的擴展性。二是可以緩解Mysql數據庫的查詢壓力，應對海量數據的復雜查詢。

ES 幾個顯著的特點，能夠有效補足 MySQL 在企業級數據操作場景的缺陷，而這也是我們將其選擇作為下游數據源重要原因。

核心特點：支持分詞檢索，多維篩選性能好，支持海量數據查詢。

• 文本搜索能力：ES 是基于倒排索引實現的搜索系統，配合多樣的分詞器，在文本模糊匹配搜索上表現得比較好，業務場景廣泛。

• 多維篩選性能好：億級規模數據使用寬表預構建(消除 join)，配合全字段索引，使 ES 在多維篩選能力上具備壓倒性優勢，而這個能力是諸如 CRM, BOSS, MIS 等企業運營系統核心訴求，加上文本搜索能力，獨此一家。

• 開源和商業并行：ES 開源生態非常活躍，具備大量的用戶群體，同時其背后也有獨立的商業公司支撐，而這讓用戶根據自身特點有了更加多樣、漸進的選擇。

ElasticSearch的發展歷史：

• 2004年Shay Banon基于Lucene開發了Compass。

• 2010年Shay Banon重寫了Compass，并取名為Elasticsearch。

后端服務系統集成ElasticSearch的好處：

• 強大的搜索功能，支持大數據量的模糊搜索功能。比如：設備、測點等。
• 一定程度上做到了讀寫分離，減輕MySQL數據庫的讀寫壓力。
• 也方便后續的ELK搭建，為服務日志提供搜索功能。
  

二、ElasticSearch基本概念

ElasticSearch官網地址：https://www.elastic.co/cn/

1、文檔和字段

ES是面向文檔（Document）存儲的，可以是數據庫中的一條商品數據，一個訂單信息。文檔數據會被序列化為json格式后存儲在elasticsearch中。

ES中的文檔和字段對應MySQL中行記錄和列屬性。

2、索引和映射

索引（Index），就是相同類型的文檔的集合。

數據庫中的表會有約束信息，用來定義表的結構、字段的名稱、類型等信息。因此，索引庫中就有映射（mapping），是索引中文檔的字段約束信息，類似表的結構約束。

ES中的索引和映射對應MySQL中的表和約束。

為避免大家對這兩個概念比較陌生，這里給出ES的索引和映射的更具象化的列子：在ES中獲取索引和映射。

• 方式1：基于HTTP請求獲取

# 獲取索引的全部信息
curl -X GET 10.0.0.101:9200/mybook?pretty=true
# 獲取索引的映射信息
curl -X GET 10.0.0.101:9200/mybook/_mapping?pretty=true

• 方式2：在kibana中基于API獲取

# 獲取索引的全部信息
GET /mybook
# 獲取索引的映射信息
GET /mybook/_mapping

無法是哪種獲取方式，都是在組裝DSL并通過Restful接口進行調用【這也是ES可被任何語言調用的原因】。
這里獲取結果如下：

# mybook索引的全部信息
{"mybook" : {"aliases" : { },"mappings" : {"dynamic" : "false","properties" : {"author" : {"type" : "text"},"id" : {"type" : "long"},"isbn" : {"type" : "text"},"publisherName" : {"type" : "text"},"title" : {"type" : "text"}}},"settings" : {"index" : {"routing" : {"allocation" : {"include" : {"_tier_preference" : "data_content"}}},"number_of_shards" : "1","provided_name" : "mybook","creation_date" : "1735287860250","number_of_replicas" : "1","uuid" : "eAT8Z7V9ThiPyRu5rnVQLQ","version" : {"created" : "7130499"}}}}
}# mybook索引的映射信息
{"mybook" : {"mappings" : {"dynamic" : "false","properties" : {"author" : {"type" : "text"},"id" : {"type" : "long"},"isbn" : {"type" : "text"},"publisherName" : {"type" : "text"},"title" : {"type" : "text"}}}}
}

• MySQL基本概念和ES基本概念對比

Elasticsearch	說明	MySQL
Index	索引(index)，就是文檔的集合，類似數據庫的表(table)	Table
Document	文檔（Document），就是一條條的數據，類似數據庫中的行（Row），文檔都是JSON格式	Row
Field	字段（Field），就是JSON文檔中的字段，類似數據庫中的列（Column）	Column
Mapping	Mapping（映射）是索引中文檔的約束，例如字段類型約束。類似數據庫的表結構（Schema）	Schema
DSL	DSL是elasticsearch提供的JSON風格的請求語句，用來操作elasticsearch，實現CRUD	SQL

3、倒排索引、文檔和詞條

倒排索引的概念比較奇特，它是根據MySQL的索引（也即正向索引）概念而命名的。

理解倒排索引需要先理解兩個基本概念：文檔和詞條。

文檔就是用來搜索的數據，每一條數據就是一個文檔。
詞條，即term，也稱詞項。就是對文檔數據或用戶搜索數據，利用某種算法分詞，得到的具備含義的詞語就是詞條。

還是來舉一個具體的例子來說明：“華為小米充電器” 就是文檔，它可以分為：華為、小米、充電器等這樣的幾個詞條。

倒排索引就是以詞條為核心，記錄哪些文檔包含了該詞條。

實際上，倒排索引除了記錄哪些文檔包含該詞條，還記錄了詞條的詞頻，詞條在文本里的偏移量等信息。

正向索引 V.S 倒排索引

• 正向索引是最傳統的，根據id索引的方式。所以當需要根據詞條查詢時，必須先逐條獲取每個文檔，然后判斷文檔中是否包含所需要的詞條，是根據文檔找詞條的過程。

• 而倒排索引則相反，是先找到用戶要搜索的詞條，根據詞條得到保護詞條的文檔的id，然后根據id獲取文檔。是根據詞條找文檔的過程。

正向索引—優點：

? 可以給多個字段創建索引；根據索引字段搜索、排序速度非常快。

正向索引—缺點：

? 根據非索引字段，或者索引字段中的部分詞條查找時，只能全表掃描。

倒排索引—優點：

? 根據詞條搜索、模糊搜索時，速度非常快。

倒排索引—缺點：

? 只能給詞條創建索引，而不是字段；無法根據字段做排序。

4、分詞器

分詞器有兩個作用：一是創建倒排索引時對文檔分詞，二是用戶搜索時，對輸入的內容進行分詞。

IK 分詞器（IK Analyzer）是專門為 ElasticSearch 設計開發的一款中文分詞插件。主要用于對中文文本進行準確且合理的分詞處理，以便更好地實現搜索、索引等功能。

IK 分詞原理：IK 分詞器基于詞典和一些特定的規則來對文本進行切分。它內部有一個較為豐富的中文詞典，里面包含了大量常見的中文詞匯、短語等內容。在對文本進行分詞時，會根據詞典中的詞條以及一些預設的規則去識別和劃分詞語。

IK分詞有兩種分詞模式：

• ik_max_word：這是一種細粒度的分詞模式。例如對于文本 “中華人民共和國”，它會被切分成 “中華”“中華人民”“中華人民共和國”“華人”“人民”“人民共和國”“共和”“共和國” 等多個詞語，盡可能多地拆分出不同組合的詞語，這種模式適用于需要更全面捕捉文本語義的場景，比如搜索引擎的索引構建，能讓更多的詞語組合參與到搜索匹配中。
• ik_smart：屬于粗粒度的分詞模式。同樣對于 “中華人民共和國”，可能就直接切分成 “中華人民共和國” 這樣一個完整的詞語，相對而言拆分得較為 “簡潔”，比較適合在一些對文本理解不需要特別細致拆分的場景中使用，例如對文檔做簡單分類等情況。

下載：可以從官方 GitHub 倉庫或者相關的軟件下載站點獲取對應的 IK 分詞器版本，要確保其版本和所使用的 ElasticSearch 版本兼容。

安裝到 ElasticSearch：將下載好的 IK 分詞器壓縮包解壓后，把整個目錄復制到 ElasticSearch 安裝目錄下的 plugins 文件夾中（如果沒有 plugins 文件夾就新建一個）。然后重啟 ElasticSearch，它就會自動加載 IK 分詞器插件。

三、ElasticSearch工作原理

相關鏈接：https://mp.weixin.qq.com/s/RUQXIyN95hvi2wM3CyPI9w

1、Term Dictionary、Term index

文檔可以通過ik分詞器分為多個詞條（又稱詞項，即Term）。詞條term會按字典排好序形成Term Dictionary（用于二分查找）。將 Term Dictionary 的部分詞條的前綴信息提取出來構建出一個精簡的目錄樹。目錄樹的節點中存放這些詞條在磁盤中的偏移量，也就是指向磁盤中的位置。這個目錄樹結構，體積小，適合放內存中，它就是所謂的 Term Index。用它可以加速搜索。

這樣當需要查找某個詞項的時候，只需要搜索 Term Index，就能快速獲得詞項在 Term Dictionary 中的大概位置。再跳轉到 Term Dictionary，通過少量的檢索，定位到詞條內容。

2、Stored Fields

倒排索引，搜索到的是文檔 id，我們還需要拿著這個 id 找到文檔內容本身，才能返回給用戶。因此還需要有個地方，存放完整的文檔內容，它就是 Stored Fields（行式存儲）。

3、Docs Values

后端的業務經常需要根據某個字段排序文檔，比如按時間排序或商品價格排序。但問題就來了，這些字段散落在文檔里。也就是說，我們需要先獲取 Stored Fields 里的文檔，再提取出內部字段進行排序。也不是說不行，但其實有更高效的做法。我們可以用空間換時間的思路，再構造一個列式存儲結構，將散落在各個文檔的某個字段，集中存放，當我們想對某個字段排序的時候，就只需要將這些集中存放的字段一次性讀取出來，就能做到針對性地進行排序。這個列式存儲結構，就是所謂的 Doc Values。

4、Segment

倒排索引用于搜索，Term Index 用于加速搜索，Stored Fields 用于存放文檔的原始信息，以及 Doc Values 用于排序和聚合。這些結構共同組成了一個復合文件，也就是所謂的"segment", 它是一個具備完整搜索功能的最小單元。

5、Lucene

多個文檔記錄可以用來生成一份 segment，如果新增文檔時，還是寫入到這份 segment，那就得同時更新 segment 內部的多個數據結構，這樣并發讀寫時性能肯定會受影響。所以規定：segment 一旦生成，則不能再被修改。如果還有新的文檔要寫入，老的segment已經寫滿，那就生成新的 segment。這樣老的 segment 只需要負責讀，寫則生成新的 segment。同時保證了讀和寫的性能。

隨著數據量增大，segment文件數會變多，這是就可以并發同時讀多個 segment。當然segment文件數也不能無限制的變多，程序會不定期的合并多個小的 segment， 也就是段合并(segment merging) 。這就是有名的Lucene，一個單機文本檢索庫。

6、高性能、高擴展性、高可用

ElasticSearch就是基于單機檢索庫Lucene構建了一個高性能、高擴展性、高可用的強大的搜索引擎。

①高性能

當多個調用方同時讀寫同一個 lucene 必然導致爭搶計算資源。 所以ES首先將不同類型的數據寫入到了不同的Lucene中，這樣在讀取數據時，根據需要搜索不同的 Index Name，這就大大降低了單個 lucene 的壓力。其次，ES還將某些Index Name內數據可能過多的單個lucene 拆成好幾份，每份都是一個 shard 分片，每個 shard 分片本質上就是一個獨立的 lucene 庫。這樣就可以將讀寫操作分攤到多個 分片 中去，大大降低了爭搶，提升了系統性能。

②高擴展性

隨著 分片 變多，如果 分片 都在同一臺機器上的話，就會導致單機 cpu 和內存過高，影響整體系統性能。

在ES中，可以使用更多的機器，將 分片 分散部署在多臺機器上，這每一臺機器，就是一個 Node。通過增加 Node 緩解機器 cpu 過高帶來的性能問題。

③高可用

高可用問題基本都是通過副本解決。ES中也是一樣， 通過給 分片 多加幾個副本。將 分片 分為 Primary shard 和 Replica shard，也就是主分片和副本分片 。主分片會將數據同步給副本分片，副本分片既可以同時提供讀操作，還能在主分片掛了的時候，升級成新的主分片讓系統保持正常運行，提高性能的同時，還保證了系統的高可用。

7、ES架構

從架構角度來看，ES給了一套方案，讓一個單機系統 lucene 變成一個高性能、高擴展、高可用的分布式系統。

在ES集群中，分為三類node角色。

• 主節點（Master Node）， 負責管理集群。
• 協調節點（Coordinate Node），負責存儲管理數據。
• 數據節點（Data Node），負責接受客戶端請求。

集群規模小的時候，一個 Node 可以同時充當多個角色，隨著集群規模變大，可以讓一個 Node 一個角色。

ES集群中的節點之間基于類似一致性算法 Raft 的方式，在節點間互相同步數據，讓所有 Node 看到的集群數據狀態都是一致的。這樣，集群內的 Node 就能參與選主過程，還能了解到集群內某個 Node 是不是掛了等信息。

• ES架構 V.S Kafka架構 V.S RocketMQ架構

確實：很多優秀的開源項目架構都是相似的。

架構	ES	Kafka	RocketMQ
消息分類	index name	topic	topic
數據分片	Shard	Partition	msgqueue
節點	node	broker	broker
高可用	多副本	多副本	master-slave
數據一致性	類似Raft協議	ISR和ack機制	Raft協議
元數據	coordinate node	zookeeper	nameserver

8、ES寫入流程

ES 對外提供 http 接口，任何語言的客戶端都可以通過 HTTP 接口接入 es，實現對數據的增刪改查。

• 當客戶端應用發起數據寫入請求，請求會先發到集群中協調節點。
• 協調節點根據 hash 路由，判斷數據該寫入到哪個數據節點里的哪個分片(Shard)，找到主分片并寫入。分片底層是 lucene，所以最終是將數據寫入到 lucene 庫里的 segment 內，將數據固化為倒排索引和 Stored Fields 以及 Doc Values 等多種結構。
• 主分片 寫入成功后會將數據同步給 副本分片。
• 副本分片 寫入完成后，主分片會響應協調節點一個 ACK，意思是寫入完成。
• 最后，協調節點響應客戶端應用寫入完成。

9、ES搜索流程

ES 的搜索流程分為兩個階段：分別是查詢階段（Query Phase）和獲取階段（Fetch Phase）

Query Phase：

• 當客戶端應用發起搜索請求，請求會先發到集群中的協調節點。
• 協調節點根據 index name 的信息，可以了解到 index name 被分為了幾個 分片，以及這些分片 分散哪個數據節點上，將請求轉發到這些數據節點的 分片 上面。
• 搜索請求到達分片后，分片 底層的 lucene 庫會并發搜索多個 segment，利用每個 segment 內部的倒排索引獲取到對應文檔 id，并結合 doc values 獲得排序信息。分片將結果聚合返回給協調節點。
• 協調節點對多個分片中拿到的數據進行一次排序聚合，舍棄大部分不需要的數據。

Fetch Phase：

• 協調節點再次拿著文檔 id 請求數據節點里的 分片，分片 底層的 lucene 庫會從 segment 內的 Stored Fields 中取出完整文檔內容，并返回給協調節點。
• 協調節點最終將數據結果返回給客戶端。完成整個搜索過程。

四、如何使用ElasticSearch

1、安裝部署

通過docker-compose一鍵部署一套es+kibana容器服務。【第七章節有詳細部署步驟】

vim docker-compose-es.yml

version: '3'services:#elasticsearch服務elasticsearch:image: docker.elastic.co/elasticsearch/elasticsearch:7.13.4container_name: elasticsearchuser: rootenvironment:- discovery.type=single-node- "ES_JAVA_OPTS=-Xms512m -Xmx512m"- TZ=Asia/Shanghaivolumes:- ./data/elasticsearch/data:/usr/share/elasticsearch/datarestart: alwaysports:- 9200:9200- 9300:9300networks:- looklook_net#查看elasticsearch數據kibana:image: docker.elastic.co/kibana/kibana:7.13.4container_name: kibanaenvironment:- elasticsearch.hosts=http://elasticsearch:9200- TZ=Asia/Shanghairestart: alwaysnetworks:- looklook_netports:- "5601:5601"depends_on:- elasticsearch
networks:looklook_net:driver: bridgeipam:config:- subnet: 172.16.0.0/16

• 一鍵啟動

docker-compose -f docker-compose-es.yml up -d

2、CRUD操作

有多種方式可以對ES進行CRUD操作，比如：shell命令行操作、kibana界面操作、基于各種開發語言的客戶端調用操作等等。這些操作的本質都一樣：即通過編寫組裝DSL語句，基于Restful請求通過http接口調用發送給ES服務端。

① index索引操作

• 查看當前節點的所有 Index

  # 如何是shell命令操作
  curl -X GET localhost:9200/_cat/indices?v
  curl -X GET http://localhost:9200/_cat/indices?v
  curl -X GET "http://localhost:9200/_cat/indices?v"
  # 等同于在kibana界面操作
  GET /_cat/indices?v
  

可以看出shell命令不大方便，所有基本都推薦使用 Kibana。后面統一使用Kibana編寫DSL的方式來演示。

• 創建Index

  PUT /mybook
  {"mappings": {"dynamic": false,  "properties": {"id": {"type": "keyword"      //使用 keyword 類型，適合精確匹配},"title": {"type": "text",        // 使用 text 類型，支持全文搜索"analyzer": "standard" // 使用標準分析器},"isbn": {"type": "text"},"author": {"type": "text"},"created_at": {"type": "date",    // 使用 date 類型，處理日期和時間"format": "yyyy-MM-dd'T'HH:mm:ss" // 日期格式},"publisherName": {"type": "text"}}}
  }
  

  常見的mapping屬性包括：

  • type：字段數據類型，常見的簡單類型有：

  • • 字符串：text（可分詞的文本）、keyword（精確值，例如：唯一id、品牌、國家、ip地址）
    • 數值：long、integer、short、byte、double、float
    • 布爾：boolean
    • 日期：date
    • 對象：object

  • index：是否創建索引，默認為true

  • analyzer：使用哪種分詞器

  • properties：該字段的子字段

• 刪除Index

  DELETE  /mybook
  

• 修改Index

  索引庫一旦創建，無法修改mapping。這是因為雖然倒排索引結構并不復雜，但是一旦索引數據結構有改變（比如改變了分詞器），就需要重新創建倒排索引。 因此索引修改只允許添加新的字段到mapping中，因為不會對倒排索引產生影響。

  PUT /mybook/_mapping
  {"properties" : {"description" : {"type" : "text"}}
  }
  

• 查詢Index

  GET /mybook
  # 等同于
  curl -X GET localhost:9200/mybook
  curl -X GET localhost:9200/mybook?pretty=true
  

② 文檔操作

索引創建完畢之后，可以開始文檔的操作。

## 創建索引
PUT /books
## 查看索引
GET /books
## 查看索引映射【這時索引映射是空的】
GET /books/_mapping## 創建文檔
POST /books/_doc
{"name": "Snow Crash liuwen","author": "Neal Stephenson","release_date": "1992-06-01","page_count": 470
}## 查看索引映射【索引映射是可以根據文檔自動生成的】
GET /books/_mappingPOST /books/_doc
{"name": "The Great Gatsby","author": "F. Scott Fitzgerald","release_date": "1925-04-10","page_count": 180,"language": "EN" 
}GET /books/_mappingPOST /_bulk
{ "index" : { "_index" : "books" } }
{"name": "Revelation Space", "author": "Alastair Reynolds", "release_date": "2000-03-15", "page_count": 585}
{ "index" : { "_index" : "books" } }
{"name": "1984", "author": "George Orwell", "release_date": "1985-06-01", "page_count": 328}
{ "index" : { "_index" : "books" } }
{"name": "Fahrenheit 451", "author": "Ray Bradbury", "release_date": "1953-10-15", "page_count": 227}
{ "index" : { "_index" : "books" } }
{"name": "Brave New World", "author": "Aldous Huxley", "release_date": "1932-06-01", "page_count": 268}
{ "index" : { "_index" : "books" } }
{"name": "The Handmaids Tale", "author": "Margaret Atwood", "release_date": "1985-06-01", "page_count": 311}## 全部查詢 
GET /books/_search
## 模糊查詢
GET books/_search
{"query": {"match": {"name": "Brave"}}
}## 支持排序、limit查詢
GET /books/_search
{"query": {"match_all": {}},"sort": [{"page_count": {"order": "desc"}}],"size": 3
}

五、MySQL數據導入到ElasticSearch的4種方案

1、同步雙寫

這是一種最為簡單的方式，在將數據寫到mysql時，同時將數據寫到ES。

• 優點：
  1、業務邏輯簡單；
  2、實時性高。

• 缺點：
  1、 硬編碼，有需要寫入mysql的地方都需要添加寫入ES的代碼；
  2、 業務強耦合；
  3、 存在雙寫失敗丟數據風險；
  4、 性能較差：本來mysql的性能不是很高，再加一個ES，系統的性能必然會下降。

• 附：
  上面說的雙寫失敗風險，包括以下幾種：
  1） ES系統不可用；
  2） 程序和ES之間的網絡故障；
  3） 程序重啟，導致系統來不及寫入ES等。
  針對這種情況，有數據強一致性要求的，就必須雙寫放到事務中來處理，而一旦用上事物，則性能下降更加明顯。

2、異步雙寫

針對多數據源寫入的場景，可以借助MQ實現異步的多源寫入，這種情況下各個源的寫入邏輯互不干擾，不會由于單個數據源寫入異常或緩慢影響其他數據源的寫入，雖然整體寫入的吞吐量增大了，但是由于MQ消費是異步消費，所以不適合實時業務場景。

• 優點：
  1、性能高；
  2、不易出現數據丟失問題，主要基于MQ消息的消費保障機制，比如ES宕機或者寫入失敗，還能重新消費MQ消息；
  3、多源寫入之間相互隔離，便于擴展更多的數據源寫入。

• 缺點：
  1、硬編碼問題，接入新的數據源需要實現新的消費者代碼；
  3、系統復雜度增加：引入了消息中間件；
  4、可能出現延時問題：MQ是異步消費模型，用戶寫入的數據不一定可以馬上看到，造成延時。

3、基于MySQL表定時同步

上面兩種方案中都存在硬編碼問題，也就是有任何對mysq進行增刪改查的地方要么植入ES代碼，要么替換為MQ代碼，代碼的侵入性太強。

如果對實時性要求不高的情況下，可以考慮用定時器來處理，具體步驟如下：
1、數據庫的相關表中增加一個字段為timestamp的字段，任何crud操作都會導致該字段的時間發生變化；
2、原來程序中的CURD操作不做任何變化；
3、增加一個定時器程序，讓該程序按一定的時間周期掃描指定的表，把該時間段內發生變化的數據提取出來；
4、逐條寫入到ES中。

典型實現案例——logstash 實現數據同步，其底層實現原理就是根據配置定期使用sql查詢新增的數據寫入ES中，實現數據的增量同步。

• 優點：
  1、不改變原來代碼，沒有侵入性、沒有硬編碼；
  2、沒有業務強耦合，不改變原來程序的性能；
  3、Worker代碼編寫簡單不需要考慮增刪改查。
• 缺點：
  1、時效性較差，由于是采用定時器根據固定頻率查詢表來同步數據，盡管將同步周期設置到秒級，也還是會存在一定時間的延遲；
  2、對數據庫有一定的輪詢壓力，一種改進方法是將輪詢放到壓力不大的從庫上。

4、基于Binlog日志實時同步

上面三種方案要么有代碼侵入，要么有硬編碼，要么有延遲，那么有沒有一種方案既能保證數據同步的實時性又沒有代入侵入呢？
當然有，可以利用mysql的binlog來進行同步。其實現原理如下：

具體步驟如下：
1） 讀取mysql的binlog日志，獲取指定表的日志信息；
2） 將讀取的信息轉為MQ；
3） 編寫一個MQ消費程序；
4） 不斷消費MQ，每消費完一條消息，將消息寫入到ES中。

• 優點：
  1、沒有代碼侵入、沒有硬編碼；
  2、原有系統不需要任何變化，沒有感知；
  3、性能高；
  4、業務解耦，不需要關注原來系統的業務邏輯。
• 缺點：
  1、構建Binlog系統復雜；
  2、如果采用MQ消費解析的binlog信息，也會像方案二一樣存在MQ延時的風險。

業界目前較為流行的實現方案——使用Canal實時同步MySQL數據到ElasticSearch。【見第六節】

總結：上述四種同步方案總結

• 1、同步雙寫是最簡單的同步方式，能最大程度保證數據同步寫入的實時性，最大的問題是代碼侵入性太強。
• 2、異步雙寫引入了消息中間件，由于MQ都是異步消費模型，所以可能出現數據同步延遲的問題。好處是在大規模消息同步時吞吐量更、高性能更好，便于接入更多的數據源，且各個數據源數據消費寫入相互隔離互不影響。
• 3、基于Mysql表定時掃描同步 ，原理是通過定時器定時掃描表中的增量數據進行數據同步，不會產生代碼侵入，但由于是定時掃描同步，所以也會存在數據同步延遲問題，典型實現是采用 Logstash 實現增量同步。
• 4、基于Binlog實時同步 ，原理是通過監聽Mysql的binlog日志進行增量同步數據。不會產生代碼侵入，數據同步的實時也能得到保障，弊端是Binlog系統都較為復雜。典型實現是采用 canal 實現數據同步。

六、Canal工作原理

Canal官網地址： https://github.com/alibaba/canal

canal [k?’n?l]，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費。

早期阿里巴巴因為杭州和美國雙機房部署，存在跨機房同步的業務需求，實現方式主要是基于業務 trigger 獲取增量變更。從 2010 年開始，業務逐步嘗試數據庫日志解析獲取增量變更進行同步，由此衍生出了大量的數據庫增量訂閱和消費業務。

基于日志增量訂閱和消費的業務包括：

• 數據庫鏡像
• 數據庫實時備份
• 索引構建和實時維護(拆分異構索引、倒排索引等)
• 業務 cache 刷新
• 帶業務邏輯的增量數據處理

當前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x。

1、MySQL主備復制原理

• MySQL master 將數據變更寫入二進制日志( binary log，其中記錄叫做二進制日志事件binary log events，可以通過 show binlog events 進行查看)。
• MySQL slave 將 master 的 binary log events 通過IO線程拷貝到它的中繼日志(relay log)。
• MySQL slave 通過SQL線程重放 relay log 中事件，將數據變更反映它自己的數據。

2、canal 工作原理

• canal 模擬 MySQL slave 的交互協議，偽裝自己為 MySQL slave，向 MySQL master 發送dump 協議。
• MySQL master 收到 dump 請求，開始推送 binary log 給 slave (即 canal)。
• Canal 解析 binary log 對象(原始為 byte 流)。
• Canal 作為 binlog 的監聽者，能夠實時捕捉到這些數據變動，然后推送到下游系統。

3、canal架構

• canal.deployer 作為基礎，負責從數據庫獲取原始的變更數據，是整個流程的數據源頭。它獲取到的數據變更事件會傳遞給后續環節。
• canal.admin 對 canal.deployer 以及整個 Canal 體系進行管理和配置，通過它可以確保 canal.deployer 能正確地連接數據源、按照合適的參數運行等；同時也可以對 canal.adapter 相關配置進行管理，比如指定 canal.adapter 要將數據同步到哪些目標端等。
• canal.adapter 依賴 canal.deployer 解析出來的數據，然后按照既定的規則和配置，將這些數據同步到目標存儲或應用中，完成數據的最終流向和落地，而它的配置和運行狀態又受到 canal.admin 的管控。

• “偵察兵” canal.deployer 是 Canal 項目的核心部署包，它主要負責與數據庫的連接以及數據變更日志（比如 MySQL 的 binlog）的讀取等基礎操作。它內部包含了處理數據抓取、解析等關鍵邏輯的模塊，是整個數據同步鏈路啟動的源頭部分。
• “控制中心” canal.admin 是 Canal 的管理端組件包。它提供了一個可視化的管理界面以及對應的管理接口，方便運維人員和開發人員對 Canal 服務進行配置管理、監控等操作。比如，通過 canal.admin 的界面，可以輕松地配置要連接的數據源信息（像數據庫的連接地址、賬號、密碼等），管理不同的 Canal 實例，查看各個實例的運行狀態（是正常運行、還是出現了錯誤等），還能進行一些動態的參數調整，像是調整數據抓取的頻率等。
• “數據搬運工” canal.adapter 主要承擔著將 canal.deployer 抓取并解析出來的數據變更內容，適配并發送到其他目標存儲或應用系統中的任務。例如，它可以把從 MySQL 數據庫解析出來的表數據變更，按照特定的格式和規則，同步到 Elasticsearch 搜索引擎中，使得 Elasticsearch 中的數據能夠及時更新，保持和MySQL 數據源的一致性；或者將數據同步到 Kafka 消息隊列，以便后續其他系統可以從 Kafka 中消費這些變更數據做進一步處理。

七、Canal實時同步Mysql數據到ElasticSearch

要將 MySQL 數據庫中表的增、刪、改操作同步到 Elasticsearch 的索引，需要正確安裝配置MySQL、Canal 、Canal Adapter 、 Elasticsearch 和 kibana。

1、MySQL安裝部署

• 安裝MySQL

參考如下鏈接，當前安裝MySQL 5.7版本。
MySQL最全安裝教程：https://blog.csdn.net/qq_41822345/article/details/117779815

• 開啟binlog日志。

vim /etc/my.cnf

## 在[mysqld]模塊添加如下內容
[mysqld]
log-bin=mysql-bin # 開啟 binlog
binlog-format=ROW # 選擇 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定義，不要和 canal 的 slaveId 重復

同步MySQL數據到ES依賴與binlog，如果MySQL沒有開啟binlog，或者binlog被刪除。那么那些沒有binlog的舊數據就沒法同步到ES了，怎么辦？——后面會提到：es提供了全量導入的方法。

• 重啟MySQL，通過show variables like ‘%XXX%’;查看上述配置是否生效。
• 準備好同步用戶。
  出于數據安全考慮，需要為canal組件單獨創建一個賬號【給予該賬號從庫權限】。

#創建用戶，密碼自己填寫，由于創建用戶時默認的密碼加密方式為caching_sha2_password，所以修改為mysql_native_password，否則服務端啟動時可能會報錯
## mysql 5.7
create user 'canal'@'%' identified with mysql_native_password by '123456';
## mysql 5.6
create user 'canal'@'%' identified by '123456';# 給新創建賬戶賦予從庫權限
grant select, replication slave, replication client on *.* to 'canal'@'%';# 刷新權限
flush privileges;

• 準備好同步數據【用于Demo演示】。

## 這里準備好一個數據庫canal，兩張表notice和result，每張表插入5條數據。
CREATE DATABASE canal;
use canal;
DROP TABLE IF EXISTS `notice`;
CREATE TABLE `notice`  (`id` int(11) NOT NULL AUTO_INCREMENT,`title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,`content` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL,`created_at` datetime NULL DEFAULT CURRENT_TIMESTAMP,`updated_at` datetime NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 24 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;INSERT INTO `notice` VALUES (1, 'Title 1', 'Content for notice 1', '2024-12-27 14:55:01', '2024-12-27 14:55:01');
INSERT INTO `notice` VALUES (2, 'Title 2', 'Content for notice 2', '2024-12-27 14:55:01', '2024-12-27 14:55:01');
INSERT INTO `notice` VALUES (3, 'Title 3', 'Content for notice 3', '2024-12-27 14:55:01', '2024-12-27 14:55:01');
INSERT INTO `notice` VALUES (4, 'Title 4', 'Content for notice 4', '2024-12-27 14:55:01', '2024-12-27 14:55:01');
INSERT INTO `notice` VALUES (5, 'Title 5', 'Content for notice 5', '2024-12-27 14:55:01', '2024-12-27 14:55:01');DROP TABLE IF EXISTS `result`;
CREATE TABLE `result`  (`id` int(11) NOT NULL AUTO_INCREMENT,`user_id` int(11) NOT NULL,`score` decimal(5, 2) NOT NULL,`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,`updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 21 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;INSERT INTO `result` VALUES (1, 1, 95.75, '2024-08-01 10:00:00', '2024-08-01 10:00:00');
INSERT INTO `result` VALUES (2, 2, 88.50, '2024-08-01 10:05:00', '2024-08-01 10:05:00');
INSERT INTO `result` VALUES (3, 3, 76.20, '2024-08-01 10:10:00', '2024-08-01 10:10:00');
INSERT INTO `result` VALUES (4, 4, 82.00, '2024-08-01 10:15:00', '2024-08-01 10:15:00');
INSERT INTO `result` VALUES (5, 5, 91.30, '2024-08-01 10:20:00', '2024-08-01 10:20:00');

2、ElasticSearch安裝部署

官網：https://www.elastic.co/cn/
ElasticSearch源碼包下載官網：https://github.com/elastic/elasticsearch
ElasticSearch安裝包下載官網：https://www.elastic.co/downloads/elasticsearch

• step0：先對操作系統做一些參數配置優化。

#1 設置打開的文件句柄數和線程數
vim /etc/security/limits.conf# 添加
# soft：軟限制；hard：硬限制
# nproc：單個用戶可打開的進程最大數
# nofile：單個進程打開文件最大數
# as：地址空間限制(unlimited：無限)
# fsize：最大文件大小
# memlock：最大鎖定內存地址空間
*               soft    nproc           65536
*               hard    nproc           65536
*               soft    nofile          65536
*               hard    nofile          65536
*               -       as              unlimited
*               -       fsize           unlimited
*               -       memlock         unlimited#2 關閉 swap 交換空間
swapoff -a && sed -i '/swap/s/^.*$/#&/' /etc/fstab#3 設置虛擬內存大小和 TCP 超時重傳次數
vim /etc/sysctl.conf# 添加
vm.max_map_count=262144
net.ipv4.tcp_retries2=5
net.core.somaxconn = 1024
vm.overcommit_memory = 1
# 默認情況下 TCP keepalive 時間為 60 秒，超時重傳 15 次。# 使上述配置生效
sysctl -p

以版本8.17為例（當前[2025年1月2日]最新版本就是8.17）。

• step1：官網下載安裝包上傳到Linux服務器

# Step1：官網下載安裝包上傳到Linux服務器
tar -xf elasticsearch-8.17.0-linux-x86_64.tar.gz -C /usr/local/useradd -u 9200 esuser
mkdir -p /data/elasticsearch/{data,logs,temp}
chown -R esuser:esuser  /data/elasticsearch/ /usr/local/elasticsearch-8.17.0/
cd /usr/local/elasticsearch-8.17.0/

• step2：修改ES的配置文件

vim config/elasticsearch.yml

## Step2：修改配置文件，添加如下內容
cluster.name: es-dev #集群名稱
node.name: es #節點名稱#######----------這個配置只需要在kibana節點安裝 ，且需要使用堆棧監測功能---------#######
node.roles: [master,data,ingest, ml, transform, remote_cluster_client]
#######----------如果使用堆棧功能，需要把安全認證關閉------------------#######path.data: /data/elasticsearch/data # 數據存儲位置
path.logs: /data/elasticsearch/logs #日志存儲位置
network.host: 0.0.0.0 #允許連接IP
# 允許跨域
http.port: 9200 # 網頁訪問端口
transport.profiles.default.port: 9300
http.cors.enabled: truehttp.cors.allow-origin: "*"
http.cors.allow-headers: "*"
#http.cors.allow-methods: "GET"
cluster.initial_master_nodes: ["es"]
action.destructive_requires_name: false
discovery.seed_hosts: ["10.0.0.101:9300"] # 集群成員#關閉安全認證配置
xpack.security.enabled: false
xpack.security.transport.ssl.enabled: false#關閉geoip配置
ingest.geoip.downloader.enabled: false
xpack.monitoring.collection.enabled: true

• step3：修改jvm配置

vim config/jvm.options

將注釋刪除，改為當前自己需要的內存大小，比如當前內存是16G，所以改為4G大小 1/4即可
-Xms4g
-Xmx4g

• step4：下載中文分詞器

./bin/elasticsearch-plugin install https://get.infini.cloud/elasticsearch/analysis-ik/8.17.0
#中途輸入 y
#下載完成會在 plugins 目錄生成analysis-ik 目錄
chown -R  esuser.esuser /usr/local/elasticsearch-8.17.0/

• step5：啟動ES

runuser -l esuser -c "/usr/local/elasticsearch-8.17.0/bin/elasticsearch -d"# 查看日志
tail -f /data/elasticsearch/logs/es-dev.log

• step6：驗證

## 查看9200端口
curl http://127.0.0.1:9200
## 輸出如下，說明啟動成功
{"name" : "es","cluster_name" : "es-dev","cluster_uuid" : "Dg6aV1E8QzetPH9vgek-zg","version" : {"number" : "8.17.0","build_flavor" : "default","build_type" : "tar","build_hash" : "2b6a7fed44faa321997703718f07ee0420804b41","build_date" : "2024-12-11T12:08:05.663969764Z","build_snapshot" : false,"lucene_version" : "9.12.0","minimum_wire_compatibility_version" : "7.17.0","minimum_index_compatibility_version" : "7.0.0"},"tagline" : "You Know, for Search"
}

3、安裝kibana

kibana源碼包下載官網：https://github.com/elastic/kibana
kibana安裝包下載官網：https://www.elastic.co/downloads/kibana

• step1：官網下載安裝包上傳到Linux服務器

tar -xf kibana-8.17.0-linux-x86_64.tar.gz -C /usr/local/
cd /usr/local/kibana-8.17.0/

• step2：修改kibana配置文件

vim config/kibana.yml

server.port: 5601server.host: "0.0.0.0"
##填本機IP或者 0.0.0.0 都可以，最好寫本機IPserver.name: "kibana-dev"
## name 名稱可以隨便指定### es集群配置
elasticsearch.hosts: ["http://127.0.0.1:9200"]pid.file: /usr/local/kibana-8.17.0/kibana.pidelasticsearch.requestTimeout: 99999i18n.locale: "zh-CN"  
#---------------------#####-------------------------
#如果高版本需要配置如下兩個參數   用戶名密碼為 elasticsearch 安全證書用戶密碼 
#如果沒有生成證書認證，可不加如下兩個參數
elasticsearch.username: "user"
elasticsearch.password: "password"

• step3：啟動kibana

useradd -u 5601 kibana
chown -R kibana:kibana /usr/local/kibana-8.17.0/
nohup /usr/local/kibana-8.17.0/bin/kibana --allow-root > /var/log/kibana.log &#查看日志
tail -f /var/log/kibana.log

• step4：驗證

## 查看5601端口 或者瀏覽器上訪問kibana頁面
curl http://127.0.0.1:5601/app/home

• 準備好同步數據。
  這里創建兩個索引：notice和result。

GET /notice/_mapping
PUT /notice
{"settings": {"number_of_shards": 1,"number_of_replicas": 1},"mappings": {"properties": {"id": {"type": "keyword"  // 使用 keyword 類型，適合精確匹配},"title": {"type": "text",    // 使用 text 類型，支持全文搜索"analyzer": "standard" // 使用標準分析器},"content": {"type": "text",    // 使用 text 類型，支持全文搜索"analyzer": "standard" // 使用標準分析器},"created_at": {"type": "date",    // 使用 date 類型，處理日期和時間"format": "yyyy-MM-dd'T'HH:mm:ss" // 日期格式},"updated_at": {"type": "date",    // 使用 date 類型，處理日期和時間"format": "yyyy-MM-dd'T'HH:mm:ss" // 日期格式}}}
}GET /result/_mapping
PUT /result
{"settings": {"number_of_shards": 1,"number_of_replicas": 1},"mappings": {"properties": {"user_id": {"type": "integer"},"score": {"type": "float"},"created_at": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"},"updated_at": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"}}}
}POST /notice/_doc/1
{"id": "1","title": "Sample Notice Title","content": "This is the content of the notice.","created_at": "2024-08-29T08:00:00","updated_at": "2024-08-29T08:00:00"
}GET /notice/_search

4、Canal安裝部署

Canal安裝包下載官網：https://github.com/alibaba/canal

• step1：官網下載安裝包上傳到Linux服務器

這里選擇版本1.1.7【當前[2025年1月2日]最新版本1.1.8屬于 α 版本，還處于內側階段】

一共三個包canal.adapter-1.1.7.tar.gz、canal.admin-1.1.7.tar.gz、canal.deployer-1.1.7.tar.gz
啟動順序：canal.admin、canal.deployer、canal.adapter

mkdir -p /usr/local/canal/canal-adapter
mkdir -p /usr/local/canal/canal-admin
mkdir -p /usr/local/canal/canal-deployer
tar -xf canal.adapter-1.1.7.tar.gz -C /usr/local/canal/canal-adapter
tar -xf canal.admin-1.1.7.tar.gz -C /usr/local/canal/canal-admin
tar -xf canal.deployer-1.1.7.tar.gz -C /usr/local/canal/canal-deployer

① canal.admin 安裝啟動

• step2：事先準備好canal_manager庫表。【sql文件在/usr/local/canal/canal-admin/conf/canal_manager.sql】

CREATE DATABASE /*!32312 IF NOT EXISTS*/ `canal_manager` /*!40100 DEFAULT CHARACTER SET utf8 COLLATE utf8_bin */;USE `canal_manager`;SET NAMES utf8;
SET FOREIGN_KEY_CHECKS = 0;-- ----------------------------
-- Table structure for canal_adapter_config
-- ----------------------------
DROP TABLE IF EXISTS `canal_adapter_config`;
CREATE TABLE `canal_adapter_config` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`category` varchar(45) NOT NULL,`name` varchar(45) NOT NULL,`status` varchar(45) DEFAULT NULL,`content` text NOT NULL,`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;-- ----------------------------
-- Table structure for canal_cluster
-- ----------------------------
DROP TABLE IF EXISTS `canal_cluster`;
CREATE TABLE `canal_cluster` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`name` varchar(63) NOT NULL,`zk_hosts` varchar(255) NOT NULL,`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;-- ----------------------------
-- Table structure for canal_config
-- ----------------------------
DROP TABLE IF EXISTS `canal_config`;
CREATE TABLE `canal_config` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`cluster_id` bigint(20) DEFAULT NULL,`server_id` bigint(20) DEFAULT NULL,`name` varchar(45) NOT NULL,`status` varchar(45) DEFAULT NULL,`content` text NOT NULL,`content_md5` varchar(128) NOT NULL,`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`),UNIQUE KEY `sid_UNIQUE` (`server_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;-- ----------------------------
-- Table structure for canal_instance_config
-- ----------------------------
DROP TABLE IF EXISTS `canal_instance_config`;
CREATE TABLE `canal_instance_config` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`cluster_id` bigint(20) DEFAULT NULL,`server_id` bigint(20) DEFAULT NULL,`name` varchar(45) NOT NULL,`status` varchar(45) DEFAULT NULL,`content` text NOT NULL,`content_md5` varchar(128) DEFAULT NULL,`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`),UNIQUE KEY `name_UNIQUE` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;-- ----------------------------
-- Table structure for canal_node_server
-- ----------------------------
DROP TABLE IF EXISTS `canal_node_server`;
CREATE TABLE `canal_node_server` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`cluster_id` bigint(20) DEFAULT NULL,`name` varchar(63) NOT NULL,`ip` varchar(63) NOT NULL,`admin_port` int(11) DEFAULT NULL,`tcp_port` int(11) DEFAULT NULL,`metric_port` int(11) DEFAULT NULL,`status` varchar(45) DEFAULT NULL,`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;-- ----------------------------
-- Table structure for canal_user
-- ----------------------------
DROP TABLE IF EXISTS `canal_user`;
CREATE TABLE `canal_user` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`username` varchar(31) NOT NULL,`password` varchar(128) NOT NULL,`name` varchar(31) NOT NULL,`roles` varchar(31) NOT NULL,`introduction` varchar(255) DEFAULT NULL,`avatar` varchar(255) DEFAULT NULL,`creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;SET FOREIGN_KEY_CHECKS = 1;-- ----------------------------
-- Records of canal_user
-- ----------------------------
BEGIN;
INSERT INTO `canal_user` VALUES (1, 'admin', '6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9', 'Canal Manager', 'admin', NULL, NULL, '2019-07-14 00:05:28');
COMMIT;SET FOREIGN_KEY_CHECKS = 1;

• step3：進入到canal.admin目錄。

cd /usr/local/canal/canal-admin

修改conf/application.yml內容如下：

server:port: 8089
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8spring.datasource:address: 127.0.0.1:3306database: canal_manager## 這里需要給canal.admin服務至少寫權限，為了簡單，直接賦予root用戶username: rootpassword: 123456driver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=falsehikari:maximum-pool-size: 30minimum-idle: 1canal:adminUser: adminadminPasswd: admin

• step4：啟動并驗證admin服務。

## 啟動
./bin/startup.sh## 觀察日志
tail -f logs/admin.log## 訪問頁面【瀏覽器訪問】
curl http://127.0.0.1:8089
# 登錄：admin/123456

② canal.deployer安裝啟動

• ste5：進入到canal.deployer目錄。

cd /usr/local/canal/canal-deployer
## 先將canal.properties備份【這里面配置太多了，實際上不需要動，放一邊去】
mv conf/canal.properties conf/canal.properties.bak
## 再修改canal_local文件【這里面配置就少了很多】
mv conf/canal_local.properties conf/canal.properties

再修改conf/canal.properties內容如下：

# register ip
canal.register.ip = 127.0.0.1
# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name =

• step6：修改conf/example/instance.properties配置

#被同步的mysql地址
canal.instance.master.address=127.0.0.1:3306
#數據庫從庫權限賬號
canal.instance.dbUsername=canal
#數據庫從庫權限賬號的密碼
canal.instance.dbPassword=123456
#數據庫連接編碼 
canal.instance.connectionCharset = UTF-8 
#需要訂閱binlog的表過濾正則表達式
canal.instance.filter.regex=.*\\..*
#這里與當前文件夾名保持一致，后面會用到
canal.mq.topic=example

• step7：上述修改不會生效，需要在canal-admin上的instance管理處進行instance的配置。如下：

• step8：啟動并驗證

## 啟動
./bin/startup.sh## 觀察日志
tail -f logs/canal/canal.log
tail -f logs/example/example.log## 訪問頁面【瀏覽器訪問】
curl http://127.0.0.1:8089
# 登錄：admin/123456
# 可以看見server管理出于啟動狀態

③ canal.adapter安裝啟動

• step9：進入到canal.adapter目錄。

cd /usr/local/canal/canal-adapter

修改conf/application.yml文件內容如下：

## 源端配置
srcDataSources:defaultDS:url: jdbc:mysql://127.0.0.1:3306/canal?useUnicode=trueusername: canalpassword: 123456
## 目標端配置
canalAdapters:- instance: example # canal instance Name or mq topic namegroups:- groupId: g1outerAdapters:- name: logger## ...- name: es8  ## 對應es8目錄## 注意，要加上http://hosts: http://127.0.0.1:9200 # or http://127.0.0.1:9300properties:mode: rest # transport for 9300， rest for 9200# security.auth: test:123456 #  only used for rest modecluster.name: es-dev   # 部署es時定義es集群名稱

修改conf/bootstrap.yml文件內容如下：

canal:manager:jdbc:url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8username: rootpassword: 123456

• step10：增加mysql表到es索引的映射配置。
  conf目錄下有es6，es7，es8三個目錄，根據es版本選擇一個目錄。一張MySQL表對應一個配置文件【也可以多張表對應一個配置文件，后面會提到】。

## 先刪除es8目標下的默認配置文件。
rm -rf conf/es8/*
## 新建notice表對應的配置文件
vim conf/es8/esMappingNotice.yml
## 文件內容如下：
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:_index: notice_id: _id_type: _docupsert: truesql: "
SELECTc.id AS _id,c.title AS title,c.content AS content,DATE_FORMAT (c.created_at, '%Y-%m-%dT%H:%i:%s') AS created,DATE_FORMAT (c.updated_at, '%Y-%m-%dT%H:%i:%s') AS updated
FROMnotice AS c
"commitBatch: 3000## 新建result表對應的配置文件
vim conf/es8/esMappingResult.yml
## 文件內容如下：
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:_index: result_id: _id_type: _docupsert: truesql: "
SELECTc.id AS _id,c.user_id AS userid,c.score AS score,DATE_FORMAT (c.created_at, '%Y-%m-%dT%H:%i:%s') AS created,DATE_FORMAT (c.updated_at, '%Y-%m-%dT%H:%i:%s') AS updated
FROMresult AS c
"commitBatch: 3000

時間類型的表結構想要存到es中必須自定義轉換器或格式化程序，將 Timestamp 轉換為 Elasticsearch 支持的日期格式 , 否則導入時會報錯。

• step11：啟動并驗證。

## 啟動執行
cd /usr/local/canal/canal-adapter
chmod 777 -R conf/es8## 啟動
./bin/startup.sh## 觀察日志
tail -f logs/adapter/adapter.log

5、整體驗證

• step12：全量導入驗證

## 執行如下命令會進行全量導入【該操作是冪等的】
curl "localhost:8081/etl/es8/esMappingNotice.yml" -X POST
curl "localhost:8081/etl/es8/esMappingResult.yml" -X POST## 輸出：{"succeeded":true,"resultMessage":"導入ES 數據：9 條"}
## 輸出：{"succeeded":true,"resultMessage":"導入ES 數據：5 條"}

• step13：增量導入驗證

## 在MySQL中執行
INSERT INTO notice (id, title, content, created_at, updated_at) VALUES (30, 'New Notice', 'This is a new notice', NOW(), NOW());## 查看日志
tail -f logs/adapter/adapter.log
## 日志輸出：2025-01-03 16:58:36.800 [pool-3-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":30,"title":"New Notice","content":"This is a new notice","created_at":1735894716000,"updated_at":1735894716000}],"database":"canal","destination":"example","es":1735894716000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"notice","ts":1735894716798,"type":"INSERT"}
## 2025-01-03 16:58:36.815 [pool-3-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.service.ESSyncService - DML: {"data":[{"id":30,"title":"New Notice","content":"This is a new notice","created_at":1735894716000,"updated_at":1735894716000}],"database":"canal","destination":"example","es":1735894716000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"notice","ts":1735894716798,"type":"INSERT"} 
## Affected indexes: notice 

• step14：查看ES中的結果

curl -X GET localhost:9200/result/_search?pretty
curl -X GET localhost:9200/notice/_search?pretty

6、啟動腳本整理

# 啟動ES
cd /usr/local/elasticsearch-8.17.0/
runuser -l esuser -c "/usr/local/elasticsearch-8.17.0/bin/elasticsearch -d"
# 啟動kibana
cd /usr/local/kibana-8.17.0/
nohup /usr/local/kibana-8.17.0/bin/kibana --allow-root > /var/log/kibana.log &# 啟動canal-admin
cd /usr/local/canal/canal-admin
./bin/stop.sh
./bin/startup.sh
# 啟動canal-deployer
cd /usr/local/canal/canal-deployer
./bin/stop.sh
./bin/startup.sh
# 啟動canal-adapter
cd /usr/local/canal/canal-adapter
./bin/stop.sh
./bin/startup.sh## 驗證
curl "localhost:8081/etl/es8/esMappingNotice.yml" -X POST
curl "localhost:8081/etl/es8/esMappingResult.yml" -X POST
curl "localhost:8081/etl/es8/esMappingPatrolPlan.yml" -X POST

八、Go-Zero項目集成ElasticSearch

1、數據同步

• canal開啟前的數據如何同步

canal-adapter提供一個REST接口可全量同步數據到ES，調用Client-Adapter服務的方法觸發同步任務。此時，canal會先中止增量數據傳輸，然后同步全量數據。待全量數據同步完成后，canal會自動進行增量數據同步。

注意：如果數據是binlog開啟前存在，則不可以使用此種方式

## eg：執行如下命令會進行全量導入【該操作是冪等的】
curl "localhost:8081/etl/es8/esMappingNotice.yml" -X POST

• binlog未開啟前的歷史數據如何同步？

因為canal是基于binlog實現全量同步的，那么未開啟binlog之前的歷史數據就無法被同步，這時需要將數據庫中的數據導出再重新導入一遍，這樣就可以生成binlog 。

2、多表連表查詢

• 多張表數據同步到一個索引中

在MySQL導入ES的過程中，一張MySQL表對應ES的一個索引index，只需要配置一個對應的yml文件。多張表如何對應到一個index索引里？在我們的業務代碼中，一個查詢接口經常需要進行多張表的連表查詢才能獲取最終的結果，這種MySQL表數據如何導入到ES中？舉個例子如下：

## 這種SQL語句查出來的數據如何導入到ES中？
SELECT `dcom_fo_patrol_plan`.*,`dcom_fo_patrol_template`.`template_name` FROM `dcom_fo_patrol_plan` LEFT JOIN `dcom_fo_patrol_template` ON `dcom_fo_patrol_template`.`id` = `dcom_fo_patrol_plan`.`template_id` WHERE `dcom_fo_patrol_plan`.`module_gid` = '26' AND `dcom_fo_patrol_plan`.`deleted_at` IS NULL ORDER BY `dcom_fo_patrol_plan`.`created_time` DESC LIMIT 10;

注意：在yml映射文件中，主表一定要在最左側，從表的數據改變也會自動同步到es中！
示例：dcom_fo_patrol_template 表中的數據改變，也會自動同步到dcom_fo_patrol_plan表對應的es索引patrolplan中。

## 先在ES中創建好對應的索引patrolplan
PUT /patrolplan
{"mappings": {"properties": {"planName": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"templateId": {"type": "keyword"},"moduleGid": {"type": "keyword"},"areaGid": {"type": "keyword"},"period": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"firstTime": {"type": "date"},"lastTime": {"type": "date"},"execTime": {"type": "date"},"createdBy": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"executor": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"executorGroup": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"status": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"executorType": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"realExecutor": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"realExecutorIds": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"uuid": {"type": "keyword"},"reportReceiver": {"type": "text","fields": {"keyword": {"type": "keyword"}}},"deletedAt": {"type": "date"},"updatedTime": {"type": "date"},"createdTime": {"type": "date"},"templateName": {"type": "text","fields": {"keyword": {"type": "keyword"}}}}}
}## 增加mysql表到es索引的映射配置
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:_index: patrolplan_id: _id_type: _docupsert: truesql: "
SELECTpp.id AS _id,pp.plan_name AS planName,pp.template_id AS templateId,pp.module_gid AS moduleGid,pp.area_gid AS areaGid,pp.period AS period,pp.first_time AS firstTime,pp.last_time AS lastTime,pp.exec_time AS execTime,pp.created_by AS createdBy,pp.executor AS executor,pp.executor_group AS executorGroup,pp.status AS status,pp.executor_type AS executorType,pp.real_executor AS realExecutor,pp.real_executor_ids AS realExecutorIds,pp.uuid AS uuid,pp.report_receiver AS reportReceiver,pp.deleted_at AS deletedAt,pp.updated_time AS updatedTime,pp.created_time AS createdTime,pt.template_name AS templateName
FROM dcom_fo_patrol_plan pp
LEFT JOINdcom_fo_patrol_template pt
ON pt.id=pp.template_id"commitBatch: 3000# 在MySQL里修改從表
update dcom_fo_patrol_template set template_name = "test_template" where id = 10;# 查看canal的adapter日志
tail -f logs/adapter/adapter.log
# 日志內容如下：
... ...
2025-01-08 14:46:12.325 [pool-3-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":10,"template_name":"test_template","bui_name":"","created_by":"dczhiwei","module_gid":"26","area_gid":"1171","created_u_id":"240","duration":1.0,"patrol_type":3,"uuid":0,"report_mail":"","group_id":0,"is_delete":0,"deleted_at":null,"update_time":1733451716000,"created_time":1736318771000}],"database":"canal","destination":"example","es":1736318771000,"groupId":"g1","isDdl":false,"old":[{"template_name":"123456","created_time":1736316855000}],"pkNames":["id"],"sql":"","table":"dcom_fo_patrol_template","ts":1736318772316,"type":"UPDATE"}
2025-01-08 14:46:12.343 [pool-3-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.service.ESSyncService - DML: {"data":[{"id":10,"template_name":"test_template","bui_name":"","created_by":"dczhiwei","module_gid":"26","area_gid":"1171","created_u_id":"240","duration":1.0,"patrol_type":3,"uuid":0,"report_mail":"","group_id":0,"is_delete":0,"deleted_at":null,"update_time":1733451716000,"created_time":1736318771000}],"database":"canal","destination":"example","es":1736318771000,"groupId":"g1","isDdl":false,"old":[{"template_name":"123456","created_time":1736316855000}],"pkNames":["id"],"sql":"","table":"dcom_fo_patrol_template","ts":1736318772316,"type":"UPDATE"} 
Affected indexes: patrolplan
... ...# 查看ES中的搜索結果
GET /patrolplan/_search
{"query": {"match": {"templateName": "test_template"}}
}
# 對比MySQL搜索的結果
SELECT `dcom_fo_patrol_plan`.*,`dcom_fo_patrol_template`.`template_name` FROM `dcom_fo_patrol_plan` LEFT JOIN `dcom_fo_patrol_template` ON `dcom_fo_patrol_template`.`id` = `dcom_fo_patrol_plan`.`template_id` WHEREtemplate_name="test_template";

3、Go-Zero框架集成ES

go-elasticsearch官網：https://github.com/elastic/go-elasticsearch

• 基于go-elasticsearch構建es-client客戶端。

package esimport ("net""net/http""strings""time"es7 "github.com/elastic/go-elasticsearch/v7""github.com/elastic/go-elasticsearch/v7/estransport""github.com/zeromicro/go-zero/core/trace""go.opentelemetry.io/otel""go.opentelemetry.io/otel/codes""go.opentelemetry.io/otel/propagation"semconv "go.opentelemetry.io/otel/semconv/v1.4.0"oteltrace "go.opentelemetry.io/otel/trace"
)type (Config struct {Addresses       []stringUsername        stringPassword        stringMaxRetries      intMaxIdleConns    int           // 全局最大空閑連接數MaxConnsPerHost int           // 每主機最大連接數IdleConnTimeout time.Duration // 空閑連接超時時間}Es struct {*es7.Client}// esTransport is a transport for elasticsearch clientesTransport struct {baseTransport *http.Transport}
)func (t *esTransport) RoundTrip(req *http.Request) (resp *http.Response, err error) {var (ctx  = req.Context()span oteltrace.Span//startTime  = time.Now()propagator = otel.GetTextMapPropagator()indexName  = strings.Split(req.URL.RequestURI(), "/")[1]tracer     = trace.TracerFromContext(ctx))ctx, span = tracer.Start(ctx,req.URL.Path,oteltrace.WithSpanKind(oteltrace.SpanKindClient),oteltrace.WithAttributes(semconv.HTTPClientAttributesFromHTTPRequest(req)...),)defer func() {//metric//metricClientReqDur.Observe(time.Since(startTime).Milliseconds(), indexName)//metricClientReqErrTotal.Inc(indexName, strconv.FormatBool(err != nil))span.End()}()req = req.WithContext(ctx)propagator.Inject(ctx, propagation.HeaderCarrier(req.Header))// 調用基礎 Transport 執行請求resp, err = t.baseTransport.RoundTrip(req)if err != nil {span.RecordError(err)span.SetStatus(codes.Error, err.Error())return}span.SetAttributes(semconv.DBSQLTableKey.String(indexName))span.SetAttributes(semconv.HTTPAttributesFromHTTPStatusCode(resp.StatusCode)...)span.SetStatus(semconv.SpanStatusFromHTTPStatusCodeAndSpanKind(resp.StatusCode, oteltrace.SpanKindClient))return
}func NewEs(conf *Config) (*Es, error) {transport := &http.Transport{MaxIdleConns:        conf.MaxIdleConns,    // 全局最大空閑連接數MaxIdleConnsPerHost: conf.MaxConnsPerHost, // 每主機最大空閑連接數MaxConnsPerHost:     conf.MaxConnsPerHost, // 每主機最大連接數IdleConnTimeout:     conf.IdleConnTimeout, // 空閑連接超時時間DialContext: (&net.Dialer{Timeout:   3 * time.Second, // 建立連接超時時間KeepAlive: time.Hour,       // 保持活動連接的時間}).DialContext,TLSHandshakeTimeout: 10 * time.Second, // TLS 握手超時時間}// 自定義連接池函數// 作用// 1. 多節點請求分發// 2. 負載均衡// 3. 故障節點管理customConnectionPoolFunc := func(addrs []*estransport.Connection, selector estransport.Selector) estransport.ConnectionPool {// 使用 RoundRobinConnectionPool（輪詢連接池）cp, err := estransport.NewConnectionPool(addrs, selector)if err != nil {panic(err)}return cp}c := es7.Config{Addresses:          conf.Addresses,Username:           conf.Username,Password:           conf.Password,MaxRetries:         conf.MaxRetries,Transport:          &esTransport{baseTransport: transport},ConnectionPoolFunc: customConnectionPoolFunc,}client, err := es7.NewClient(c)if err != nil {return nil, err}return &Es{Client: client,}, nil
}func MustNewEs(conf *Config) *Es {es, err := NewEs(conf)if err != nil {panic(err)}return es
}

• Demo

package esimport ("context""strings""testing"
)func TestElastic(t *testing.T) {esClient := MustNewEs(&Config{Addresses: []string{"http://10.0.0.101:9200"},Username:  "es",Password:  "123456",})searchResult, err := esClient.Search(esClient.Search.WithContext(context.Background()),esClient.Search.WithIndex("mybook"),esClient.Search.WithBody(strings.NewReader(`{"query":{"match":{"title":"三體"}}}`)),esClient.Search.WithPretty(),)if err != nil {t.Fatal(err)}t.Log("searchResult: ", searchResult)
}