百萬級商品數(shù)據實時同步搜索系統(tǒng)主要面臨以下幾個難題：
商家數(shù)據庫和商品數(shù)據庫是多臺不同的服務器，并且數(shù)據量達百萬級，如何才能實現(xiàn)跨數(shù)據庫的數(shù)據同步呢？
商家和商品的數(shù)據是有從屬關系的，不然就會把肯德基的香辣雞腿堡掛到麥當勞去，這就尷尬了！
商家商品數(shù)據是經常更新的，比如修改價格、庫存、上下架等，那搜索服務可不能搜出一堆過時的數(shù)據，如果客戶明明搜出來的商品，點進去后卻已下架了，那么客戶就要吐槽了！如何實現(xiàn)搜索數(shù)據與源數(shù)據庫增刪改均實時同步呢？
帶著以上三個問題，我們開始了搜索服務的整體架構設計。
系統(tǒng)架構設計思路
為了設計出合適的系統(tǒng)架構，我們分析了現(xiàn)狀：
首先，商家數(shù)據和商品數(shù)據分別存儲在 2 個獨立的 MySQL8 數(shù)據庫，為滿足商家數(shù)據和商品數(shù)據的關聯(lián)，我們需要將兩個庫中所需要的表實時 ETL 到我們的搜索系統(tǒng)數(shù)據庫。
其次，數(shù)據從商家、商品數(shù)據庫 ETL 到搜索系統(tǒng)數(shù)據庫后，需要實時的組合成為商家關聯(lián)商品數(shù)據結構，并以父子文檔的格式，存儲到 ES 中。
最后，商家、商品數(shù)據庫的增刪改操作，需要實時的同步到 ES 中，也就是 ES 中的數(shù)據，需要支持實時的增加、刪除和修改。
為此，我們設計了 2 個 Canal 組件，第一個 Canal 實現(xiàn)數(shù)據 ETL，把商家、商品數(shù)據庫的某些表及字段，抽取到搜索服務數(shù)據庫。
再利用第二個 Canal，讀取搜索服務 MySQL 數(shù)據庫的 Binlog，實時傳輸?shù)?Kafka 消息隊列，再由 canal adapter 對數(shù)據進行關聯(lián)、父子文檔映射等，將處理好的數(shù)據存儲到 ElasticSearch 中。
具體系統(tǒng)架構設計如下圖所示：商家商品搜索系統(tǒng)架構設計
項目實戰(zhàn)
環(huán)境及軟件說明
操作系統(tǒng)：CentOS 7
canal：canal.adapter-1.1.4，canal.deployer-1.1.4
kafka：kafka_2.12-2.3.0
ElasticSearch：elasticsearch-6.3.2
kibana：kibana-6.3.2
用 Canal 實現(xiàn)數(shù)據 ETL 到 MySQL8
這個步驟是利用 Canal 從 2 個獨立的 MySQL8 數(shù)據庫中，抽取需要的表到搜索服務的 MySQL 數(shù)據庫。
①安裝 canaldeployer
解壓 canal.deployer-1.1.4.tar.gz，并配置 canal deployer。
進入 canaldeployer/conf 目錄，修改 canal.properties 文件，主要配置 serverMode、MQ 和 destination 三部分。
首先，我們 serverMode 修改為 Kafka 模式，增加系統(tǒng)緩沖能力以及提高系統(tǒng)穩(wěn)定性：接著，配置 Kafka 的 MQ 信息（Kafka 請自行安裝）：
最后，配置需要實例化的 instance，這里配置了 3 個，表示 canal deploy 會啟動這 3 個實例，同步 MySQL 的 Binlog 到 Kafka 的 Topic 內。
如下圖所示：配置 canal deployer instance：進入 canaldeployer/conf/example 目錄，發(fā)現(xiàn)有一個instance.properties 文件，這是 Canal 給的示例，我們可以參考其配置。
我們拷貝整個 example 目錄，并重命名為上個步驟配置的 destination 之一，如 xxxsearch。
進入 xxxsearch 目錄，編輯 instance.properties 文件，主要配置源數(shù)據庫信息、所需數(shù)據表及字段，以及指定 Kafka 的 Topic 名。
這樣源數(shù)據庫的 Binlog 就會轉換為 Json 數(shù)據，并實時的通過 canal deployer 傳輸?shù)?Kafka 該 Topic 中。
如下所示：
進入 canaldeployer/bin 目錄，執(zhí)行 ./startup.sh，啟動 canal deployer 及所屬實例。至此 canal deployer 搭建完成。
②安裝 canal.adapter
我們需要利用 canal.adapter 將 Kafka Topic 中的 binlog json 數(shù)據，經過清洗轉換等操作，存儲到 MySQL8 中。由于 Canal 原生是不支持 MySQL8 的，故我們需要做一些調整。
增加 MySQL8 連接驅動：解壓 canal.adapter-1.1.4.tar.gz，進入 canaladapter/lib 目錄，移除 mysql-connector-java-5.1.40.jar，導入 mysql-connector-java-8.0.18.jar。
配置 canal adapter，使數(shù)據輸出到 MySQL8：進入 canaladapter/conf 目錄，編輯 application.yml 文件，主要配置消費 Kafka、源數(shù)據庫信息和搜索系統(tǒng)數(shù)據庫信息。
如下所示：
接著，進入 canaladapter/conf/rdb 目錄，以官方提供的 mytest_user.yml 為例，配置 Kafka Topic 名、源數(shù)據庫名、源數(shù)據表名，以及目標數(shù)據庫名和目標數(shù)據表名，建議一張表對應一個 yml 文件。
啟動 canaladapter：進入canaladapter/bin 目錄，執(zhí)行 ./startup.sh，啟動 canal adapter，觀察 logs/adapter/adapter.log 日志文件，手動在搜索系統(tǒng)數(shù)據庫新增一條記錄，看是否會打印如下日志，即有 2 條記錄，一條 INFO，一條 DEBUG，則表示配置成功。
至此，數(shù)據 ETL 階段搭建完成，數(shù)據可從兩個不同的 MySQL8 數(shù)據庫，實時同步到搜索服務的 MySQL 數(shù)據庫。
實現(xiàn)數(shù)據多表關聯(lián)、父子文檔映射
①配置第二個 Canal 的 canaladapter
進入 canaladapter/conf 目錄，編輯 application.yml 文件，主要配置消費 Kafka、搜索系統(tǒng)數(shù)據庫，和 ES 連接信息。
②配置多表關聯(lián)
進入 canaladapter/conf/es 目錄，vim mytest_user.yml，編輯多表關聯(lián)配置：
注意，sql支持多表關聯(lián)自由組合, 但是有一定的限制：
主表不能為子查詢語句。
只能使用
left outer join即最左表一定要是主表。關聯(lián)從表如果是子查詢不能有多張表。
主
sql中不能有where查詢條件（從表子查詢中可以有where條件但是不推薦, 可能會造成數(shù)據同步的不一致，比如修改了where條件中的字段內容）。關聯(lián)條件只允許主外鍵的'='操作不能出現(xiàn)其他常量判斷比如：
on a.role_id=b.id and b.statues=1。關聯(lián)條件必須要有一個字段出現(xiàn)在主查詢語句中比如：
on a.role_id=b.id其中的a.role_id或者b.id必須出現(xiàn)在主select語句中。-
ElasticSearch的mapping屬性與 sql 的查詢值將一一對應（不支持select *）。比如：
select a.id as _id，a.name，a.email as _email from user，其中name將映射到es mapping的name field，_email將映射到mapping的_email field，這里以別名（如果有別名）作為最終的映射字段。這里的_id可以填寫到配置文件的_id: _id映射。
③配置父子文檔
以官方的 biz_order.yml 為例，vim biz_order.yml，配置父子文檔映射：
④在 ElasticSearch6 中，建立 index 和父子文檔映射關系
進入 Kibana 頁面，點擊 Dev Tools，執(zhí)行如下命令，即可建立索引及父子文檔映射：
其中，ES6 和 Kibana 的安裝，在此無特別配置，不做贅述。
⑤啟動 canal adapter
進入 canaladapter/bin 目錄，執(zhí)行 ./startup.sh，啟動 canal adapter，觀察 logs/adapter/adapter.log 日志文件，手動在搜索系統(tǒng)數(shù)據庫新增一條記錄，看是否會打印如下日志，如打印則表示配置成功。
運行結果
現(xiàn)在，我們可以通過 Kibana 來執(zhí)行 DSL 語句來查詢看看。
我們事先已在商家系統(tǒng)中增加了一個“肯德基”商店，然后在商品系統(tǒng)中添加了“西紅柿”和“新鮮西紅柿”2 個商品，并將商品關聯(lián)到“肯德基”上。
接著我們查詢“肯德基”或者“西紅柿”，得到以下是查詢的結果（去除了 ES 默認字段）：
由圖可見，我們可以通過商家名查詢商品，也可通過商品名查詢商店和商品，并且 Canal 支持數(shù)據的實時增刪改，所以 ES 的數(shù)據也會與商家系統(tǒng)和商品系統(tǒng)保持一致，同時數(shù)據結構包含商家及對應的商品，滿足業(yè)務需求。
總結
至此，基于 Canal、Kafka、MySQL8、ElasticSearch6 技術的商家商品搜索系統(tǒng)基礎框架搭建完成。
我們采用 canal deployer 實時讀取商家、商品系統(tǒng)的 MySQL 數(shù)據庫 Binlog，并發(fā)送至 Kafka。
接著由 canal adapter 消費 Kafka，并將 binlog json 數(shù)據進行多表關聯(lián)、父子文檔映射，最后存儲到 ES6 中，供上層搜索服務調用。
