基于Zookeeper構建準實時索引更新系統及其監控

鄧杰 童孟軍 胡文澤 林英杰 胡燚

摘? 要： Mysql數據庫表切換過程，需要保證兩個相同表同時寫入。Solr作為企業級的搜索框架，其自帶的索引更新系統是增量更新，實時性差。為了滿足互聯網企業對數據的實時性要求，設計和實現了一款基于Zookeeper的能夠準實時更新Solr索引和實現數據庫雙寫的系統。用戶需要提交配置信息到系統，當Mysql數據源發生內容變更時，系統就能夠實時捕獲，將變更內容經過數據轉化并實時同步更新Solr索引。

關鍵詞： Zookeeper; Solr; Mysql; 實時同步; 分布式

中圖分類號：TP392? ? ? ? ? 文獻標識碼：? ? ?文章編號：1006-8228（2020）02-58-04

Quasi–real-time index update system using Zookeeper and its monitoring

Deng Jie， Tong Mengjun， Hu Wenze， Lin Yingjie， Hu Yi

（College of Information Engineering， Zhejiang A&F University， Hangzhou， Zhejiang， 311300， China）

Abstract： Mysql database table switching process， it is needed to ensure that two same tables are written at the same time. As an enterprise-class search framework， Solr's built-in index update system is incremental update with poor real-time performance. In order to meet the real-time requirements of Internet enterprises， a system is designed and implemented by using Zookeeper， which can quasi–real-timely update Solr index and realize database dual write. Users need to submit configuration information to the system. When the content of Mysql data source changes， the system can capture the changed content in real time， and update Solr index synchronously in real time through the data transformation.

Key words： Zookeeper; Solr; Mysql; real-time; synchronization

0 引言

當前互聯網企業對數據的實時性要求越來越高。所以本文研究的內容著眼于實時的數據同步，整個系統分為數據抓取分發平臺部分和索引更新部分。基于這個系統來解決下面兩方面的問題。

⑴ 很多公司會將部分數據冗余存在搜索平臺Solr上，當用戶請求數據時，直接從搜索平臺Solr上獲取而不是訪問數據庫（因為數據庫不支持復雜的搜索邏輯）。所以需要一款能夠實時更新搜索索引的系統，當數據庫內容發生變更時，及時更新對應的Solr上的索引。

⑵ 有些特殊情況需要替換線上運行的數據庫mysql內的某個表，但又因為是線上運行的數據庫，不應該讓業務感知到這種變更，所以需要一個系統進行漸進的替換過程。

1 核心框架和相關技術介紹

1.1 系統運用到的核心框架

zookeeper提供分布式協調服務，提供諸如統一命名服務、配置管理和分布式鎖、分布式消息等分布式的基礎服務。它是一個典型的分布式數據一致性的解決方案，分布式應用可以基于zookeeper實現發布/訂閱、負載均衡、命名服務、分布式協調/通知、集群管理/master選舉、分布式鎖和分布式隊列等功能[1]。功能強大又好用。

kafka是一款基于發布訂閱模型的消息隊列。它是一個能夠提供實時數據傳輸的平臺，具有高吞吐、低延遲的特點。使用它的原因：一是可以處理更多的消息，不受單臺服務器的限制，二是分區可作為并行處理的單元。這樣使所有的請求給多臺服務器處理[2]。

Kafka Connect是一種用于Kafka和其他數據系統之間進行數據傳輸的工具。僅關注數據的復制，不處理其他任務，對數據的傳輸進行管理和監控。

Debezium是Kafka Connect的一種實現，主要用于數據庫和kafka之間的數據傳輸。它是一個CDC（Change Data Capture）系統，能實時捕獲上游數據的變動。然后記錄到一個或者多個Kafka topic。

solr是一款開源的企業級搜索框架，其主要功能有全文檢索，分詞，拼音檢索等。通過solr索引，能夠在短時間從海量數據里得到用戶關心的數據。

1.2 系統運用到的主要技術

⑴ Spring技術。Spring是現在非常流行的一個框架。SpringMVC是Spring的一個模塊，它提供完整的MVC模型解決方案[3]。本系統使用REST接口來管理同步任務，通過使用Spring和SpringMVC能夠很方便、快速的搭建一個后端應用，簡化開發流程[4]。

⑵ Java多線程技術。只有一個線程的系統，運行效率必定糟糕，運用到多線程技術后可以同時運行不同的同步任務，大大的提高了運行的效率[5]。

⑶ 分布式技術。本系統基于搶占式任務調度方式，保證同步任務的高可用，可以在不同機器運行。

2 系統實現

2.1 實時數據管道

實時數據管道主要數據來自mysql。當數據庫信息發生變更時，mysql內變更前和變更后的內容會被發送到實時數據管道上，使各個接入方在極短時間內收到數據庫變更信息。

實時數據管道由3個核心部分組成：CDC模塊、系統中間體Kafka和Schema registry。

圖1是一個簡單的架構圖，為了展示用戶如何通過管道得到數據變更。Debezium實時的抓取到mysql的數據變更并使用Avro將數據序列化得到schema和值，其中schema被提交到schema registry并返回id，之后id和序列化后的值一起發送到kafka。數據管道下游邏輯由開發人員實現：初始化一個KafkaConsumer，訂閱指定的Kafka Topic，根據拿到的數據schema id，從schema registry里獲取到對應的schema，然后使用Avro將拉取到的數據的值和schema反序列化成一條消息交付給用戶。本系統中的實時數據管道部分還增加了基于組件JMX信息的監控，用來實時觀察數據管道的狀況。

2.2 Solr索引和Mysql數據的同步系統SIS

SIS目的是解決Solr索引實時更新和數據庫雙寫的問題。由服務端和客戶端兩部分組成。用戶向SIS服務端提交任務，SIS客戶端從SIS服務端監聽到新任務后，啟動任務。

⑴ SIS服務端的實現

SIS服務端有提交同步任務、刪除同步任務、更新同步任務三個功能。

用戶向服務端提交任務，服務端創建/sis/task和/taskX節點。/sis/task節點為SIS同步任務根節點，/sis/task/taskX表示一個同步任務節點，其配置信息都會保存在自己內部。

新任務提交的過程的概括如下：首先SIS服務端啟動時，首先會嘗試向zookeeper注冊/sis/task持久節點。然后用戶提交一份同步任務的配置內容到SIS服務端。SIS服務端收到任務創建請求后，向zk創建/sis/task/taskX 持久節點，其中taskX為同步任務的名稱。同時，同步任務的配置信息會被保存到/sis/task/taskX節點內。Zookeeper提供的分布式協調功能對同一個節點的多個創建請求，只會有一個請求能成功，這也保證了不會有多個相同任務被創建。

⑵ SIS服務端的設計

SIS客戶端的設計圍繞zookeepe展開，它負責同步任務組件創建，調度器初始化等工作。這里的實現非常復雜，在這里簡單的闡述客戶端任務的情況。

圖2是SIS客戶端創建任務工作流程。其中Client表示SIS服務端，它們以集群的形式運行，每一個Client都是對等的。

⑴ 每個SIS客戶端啟動時，會向/sis/task節點注冊監聽器，監聽該節點子節點變化情況。當/sis/task子節點增加，刪除，內容更新時SIS客戶端會收到通知。

⑵ SIS服務端根據用戶創建任務請求，創建/sis/task/taskX 任務節點，其中taskX為同步任務的名稱，它是一個持久節點。

⑶ 所有SIS客戶端都會得到節點/sis/task/taskX被創建的消息。SIS客戶端收到回調之后，都會向/sis/task/taskX節點注冊監聽器。Zookeeper保證只會有一個客戶端請求成功，開始任務同時將同步任務的運行狀態寫入到lock節點內部。

⑷ 如果此時運行同步任務taskX的服務器發生宕機， 那么SIS和zookeeper的連接將會斷開，并且lock臨時節點將自動刪除。剩余的SIS客戶端由于添加了對同步任務的監聽器會收到同步任務中斷的通知，又開始對/sis/task/taskX任務節點加鎖。加鎖成功的SIS客戶端，從任務節點讀取配置信息，重新啟動同步任務。基于這個機制實現SIS同步任務的高可用。

⑸ 如果用戶主動提交刪除任務請求，那么SIS服務端首先將/sis/task/taskX/lock鎖節點的狀態信息更新為WAIT_FOR_CLOSE，這表示該任務節點等待刪除，隨后刪除鎖節點和/sis/task/taskX任務節點，之后所有在/sis/task/taskX任務節點注冊監聽器的SIS客戶端都會收到鎖節點被用戶主動刪除的通知，但都不做任何響應。任務節點刪除之后，所有向/sis/task節點注冊監聽器的SIS客戶端收到任務節點被刪除的通知，SIS客戶端根據通知內容判斷同步任務是否運行在自己所在服務器來同步任務和清理資源。

⑹ 如果用戶主動提交更新同步任務配置信息請求，那么SIS服務端會更新/sis/task/taskX的節點內容。隨后所有在/sis/task節點注冊監聽器的SIS客戶端都會收到通知，并根據通知獲取到具體哪個任務節點需要更新，隨后更新/sis/task/taskX/lock鎖節點狀態為NEED_UPDATE。之后所有在/sis/task/taskX節點注冊監聽器的SIS客戶端收都會收到鎖節點內容被更新的通知，并判斷對應的同步任務是否在自己所在的服務器，如果是則再次判斷任務狀態，如果為NEED_UPDATE，那么就停止老的同步任務，清理資源，刪除同步任務下的鎖節點。鎖節點被刪除后，和第⑷步類似，創建新的同步任務。

上述六個步驟概述了SIS客戶端對任務的調度過程，基于zookeeper的SIS客戶端和服務端的實現，讓SIS同步任務能夠高可用，即使某一臺服務器宕機，同步任務也不會中斷。

2.3 實時數據管道和SIS

實時數據管道可以應用于以下的場景：數值統計、實時數據分析、響應式編程。實時數據管道能夠讓開發人員實現實時ETL（Extract-Transform-Load），提供實時、無限的數據流。

SIS可以解決solr索引更新延時大的缺點，實現索引的實時更新。并且還能夠完美解決Mysql雙寫需要開發人員在項目代碼里添加額外代碼，實現數據寫入兩個庫的問題。

如下是數據管道和SIS相結合實現mysql雙寫。

現在在同一個數據庫里有2張表，分別是userinfo和test表。本系統將實現：當userinfo表有內容變更時，test表能立刻同步。用戶提交配置內容到同步服務器，指定userinfo表的變更需要被同步到test表，點擊create創建數據庫雙寫同步任務。

提交的配置信息如圖3，指定SIS同步的數據來自數據管道userinfo相關的topic。中間有多個處理過程包括數據的冗余，轉換。最后數據會被寫入mysql的test表。

從圖4可以看到，Userinfo表里userid為1的數據變更前timestamp字段的值為null，當行內容有更新時，timestamp值會被自動更新為內容更新的時間。

更新userinfo表userid為1的行數據，將username更新為name111，如圖5所示。timestamp字段的值被更新為此行內容變更時的時間2018-05-15 23：37：03。因為之前創建了mysql雙寫同步任務，所以userinfo的變更內容，會被同步到test表。test表的username值變更為name111，并且timestamp為test表userid為1的行變更時的時間 2018-05-14 23：37：04。和userinfo的timestamp值2018-05-15 23：37：03相比，同步userinfo變更內容到test表，只花費了1秒。

2.4 對比

圖6⑴是使用本系統后的效果圖。highwater表示每個時間點mysql總共有多少變更數據，offset表示當前消費的數據量。從中可以看到，2條線是重合的，也就是說在每個時間點的mysql變更，本系統都能夠實時的處理消費。由于監控的原因，offset可能會高于highwater，offset高于highwater表示SIS消費是完全跟上了mysql變更。

圖6⑵是不使用本系統而是使用增量的方式處理消息的延時圖。可以看到下面的線offset總是經過一段時間后才上漲，而不能做到實時的和highwater保持一致。通過對比，使用了本系統后，能夠做到數據的實時處理。

3 總結

本文主要研究了一個通用的，能夠服務于不同系統的數據同步系統。通過本系統，開發人員只需要編寫一份簡單的描述文件，說明要同步的數據從哪來、到哪去，比如指定需要同步的數據是哪個mysql的表，這些數據會被同步到solr還是mysql的另一個新表等。系統根據配置內容就能自動同步。有了此系統能夠讓開發人員專注于業務開發而不需要花費大量的精力在業務之外的代碼編寫上，提高了開發效率。

參考文獻（References）：

[1] 倪超.從Paxos到Zookeeper：分布式一致性原理與實踐[M].北京：機械工業出版社，2015.

[2] 牟大恩.Kafka入門與實踐[M].人民郵電出版社，2017：59-89

[3] Craig Walls.Spring實戰（第4版） [M].人民郵電出版社，2016：187-205

[4] 明日科技.Java Web從入門到精通 [M].清華大學出版社，2012：78-89

[5] 葛一鳴，郭超.實戰Java高并發程序設計[M].電子工業出版社，2015：100-110

[6] 鳥哥.鳥哥的Linux私房菜[M].人民郵電出版社，2010：120-150

[7] 克雷格·沃斯.Spring Boot實戰[M].人民郵電出版社，2016：93-134

[8] Bruce Eckel.Java編程思想（第4版）[M].機械工業出版社，2007：135-150

[9] 瘋狂軟件.Spring+MyBatis企業應用實戰[M].電子工業出版社，2017：87-102

[10] Raoul-Gabriel Urma， Mario Fusco， Alan Mycroft.Java 8 in Action[M].USA：Manning，2014：153-160