日處理20億數據，實時用戶行爲服務系統架構實踐

攜程實時用戶行爲服務做爲基礎服務，目前廣泛應用在多個場景中，好比猜你喜歡（攜程的推薦系統）、動態廣告、用戶畫像、瀏覽歷史等等。

以猜你喜歡爲例，猜你喜歡爲應用內用戶提供潛在選項，提升成交效率。旅行是一項綜合性的需求，用戶每每須要不止一個產品。做爲一站式的旅遊服務平臺，跨業務線的推薦，特別是實時推薦，能實際知足用戶的需求，所以在上游提供打通各業務線之間的用戶行爲數據有很大的必要性。

攜程原有的實時用戶行爲系統存在一些問題，包括：1）數據覆蓋不全；2）數據輸出沒有統一格式，對衆多使用方提升了接入成本；3）日誌處理模塊是web service，比較難支持多種數據處理策略和實現方便擴容應對流量洪峯的需求等。

而近幾年旅遊市場高速增加，數據量愈來愈大，而且會持續快速增加。有愈來愈多的使用需求，對系統的實時性，穩定性也提出了更高的要求。總的來講，當前需求對系統的實時性/可用性/性能/擴展性方面都有很高的要求。
 

1、架構

 

這樣的背景下，咱們按照以下結構從新設計了系統：

圖1：實時用戶行爲系統邏輯視圖

 

新的架構下，數據有兩種流向，分別是處理流和輸出流。

在處理流，行爲日誌會從客戶端（App/Online/H5）上傳到服務端的Collector Service。Collector Service將消息發送到分佈式隊列。數據處理模塊由流計算框架完成，從分佈式隊列讀出數據，處理以後把數據寫入數據層，由分佈式緩存和數據庫集羣組成。

輸出流相對簡單，Web Service的後臺會從數據層拉取數據，並輸出給調用方，有的是內部服務調用，好比推薦系統，也有的是輸出到前臺，好比瀏覽歷史。系統實現採用的是Java+Kafka+Storm+Redis+MySQL+Tomcat+spring的技術棧。

1. Java：目前公司內部Java化的氛圍比較濃厚，而且Java有比較成熟的大數據組件

 

2. Kafka/Storm：Kafka做爲分佈式消息隊列已經在公司有比較成熟的應用，流計算框架Storm也已經落地，而且有比較好的運維支持環境。

 

3. Redis： Redis的HA，SortedSet和過時等特性比較好地知足了系統的需求。

 

4. MySQL: 做爲基礎系統，穩定性和性能也是系統的兩大指標，對比NoSQL的主要選項，好比HBase和ElasticSearch，十億數據級別上MySQL在這兩方面有更好的表現，而且通過設計可以有不錯的水平擴展能力。

 

目前系統天天處理20億左右的數據量，數據從上線到可用的時間在300毫秒左右。查詢服務天天服務8000萬左右的請求，平均延遲在6毫秒左右。下面從實時性/可用性/性能/部署幾個維度來講明系統的設計。
 

2、實時性

 

做爲一個實時系統，實時性是首要指標。線上系統面對着各類異常狀況。例如以下幾種狀況：

1.突發流量洪峯，怎麼應對；
2.出現失敗數據或故障模塊，如何保證失敗數據重試並同時保證新數據的處理；
3.環境問題或bug致使數據積壓，如何快速消解；
4.程序bug，舊數據須要從新處理，如何快速處理同時保證新數據；

 

系統從設計之初就考慮了上述狀況。

首先是用storm解決了突發流量洪峯的問題。storm具備以下特性：

圖2：Storm特性

 

做爲一個流計算框架，和早期大數據處理的批處理框架有明顯區別。批處理框架是執行完一次任務就結束運行，而流處理框架則持續運行，理論上永不中止，而且處理粒度是消息級別，所以只要系統的計算能力足夠，就能保證每條消息都能第一時間被發現並處理。

對當前系統來講，經過storm處理框架，消息能在進入kafka以後毫秒級別被處理。此外，storm具備強大的scale out能力。只要經過後臺修改worker數量參數，並重啓topology（storm的任務名稱），能夠立刻擴展計算能力，方便應對突發的流量洪峯。

對消息的處理storm支持多種數據保證策略，at least once，at most once，exactly once。對實時用戶行爲來講，首先是保證數據儘量少丟失，另外要支持包括重試和降級的多種數據處理策略，並不能發揮exactly once的優點，反而會由於事務支持下降性能，因此實時用戶行爲系統採用的at least once的策略。這種策略下消息可能會重發，因此程序處理實現了冪等支持。

storm的發佈比較簡單，上傳更新程序jar包並重啓任務便可完成一次發佈，遺憾的是沒有多版本灰度發佈的支持。

圖3：Storm架構

 

在部分狀況下數據處理須要重試，好比數據庫鏈接超時，或者沒法鏈接。鏈接超時可能立刻重試就能恢復，可是沒法鏈接通常須要更長時間等待網絡或數據庫的恢復，這種狀況下處理程序不能一直等待，不然會形成數據延遲。實時用戶行爲系統採用了雙隊列的設計來解決這個問題。

圖4：雙隊列設計

 

生產者將行爲紀錄寫入Queue1（主要保持數據新鮮），Worker從Queue1消費新鮮數據。若是發生上述異常數據，則Worker將異常數據寫入Queue2（主要保持異常數據）。

這樣Worker對Queue1的消費進度不會被異常數據影響，能夠保持消費新鮮數據。RetryWorker會監聽Queue2，消費異常數據，若是處理尚未成功，則按照必定的策略（以下圖）等待或者從新將異常數據寫入Queue2。

 

圖5：補償重試策略

 

另外，數據發生積壓的狀況下，能夠調整Worker的消費遊標，從最新的數據從新開始消費，保證最新數據獲得處理。中間未經處理的一段數據則啓動backupWorker，指定起止遊標，在消費完指定區間的數據以後，backupWorker會自動中止。（以下圖）

 

圖6：積壓數據消解

 

3、可用性

 

做爲基礎服務，對可用性的要求比通常的服務要高得多，由於下游依賴的服務多，一旦出現故障，有可能會引發級聯反應影響大量業務。項目從設計上對如下問題作了處理，保障系統的可用性：

1.系統是否有單點？
2.DB擴容/維護/故障怎麼辦？
3.Redis維護/升級補丁怎麼辦？

4.服務萬一掛了如何快速恢復？如何儘可能不影響下游應用？

 

首先是系統層面上作了全棧集羣化。kafka和storm自己比較成熟地支持集羣化運維；web服務支持了無狀態處理而且經過負載均衡實現集羣化；redis和DB方面攜程已經支持主備部署，使用過程當中若是主機發生故障，備機會自動接管服務；經過全棧集羣化保障系統沒有單點。

另外系統在部分模塊不可用時經過降級處理保障整個系統的可用性。先看看正常數據處理流程：（以下圖）

 

圖7：正常數據流程

 

在系統正常狀態下，storm會從kafka中讀取數據，分別寫入到redis和mysql中。服務從redis拉取（取不到時從db補償），輸出給客戶端。DB降級的狀況下，數據流程也隨之改變（以下圖）

 

圖8：系統降級-DB

 

當mysql不可用時，經過打開db降級開關，storm會正常寫入redis，但再也不往mysql寫入數據。數據進入reids就能夠被查詢服務使用，提供給客戶端。另外storm會把數據寫入一份到kafka的retry隊列，在mysql正常服務以後，經過關閉db降級開關，storm會消費retry隊列中的數據，從而把數據寫入到mysql中。redis和mysql的數據在降級期間會有不一致，但系統恢復正常以後會經過retry保證數據最終的一致性。redis的降級處理也相似（以下圖）

 

圖9：系統降級-Redis

 

惟一有點不一樣的是Redis的服務能力要遠超過MySQL。因此在Redis降級時系統的吞吐能力是降低的。這時咱們會監控db壓力，若是發現MySQL壓力較大，會暫時中止數據的寫入，下降MySQL的壓力，從而保證查詢服務的穩定。

爲了下降故障狀況下對下游的影響，查詢服務經過Netflix的Hystrix組件支持了熔斷模式（以下圖）。

 

圖10：Circuit Breaker Pattern

 

在該模式下，一旦服務失敗請求在給定時間內超過一個閾值，就會打開熔斷開關。在開關開啓狀況下，服務對後續請求直接返回失敗響應，不會再讓請求通過業務模塊處理，從而避免服務器進一步增長壓力引發雪崩，也不會由於響應時間延長拖累調用方。

開關打開以後會開始計時，timeout後會進入Half Open的狀態，在該狀態下會容許一個請求經過，進入業務處理模塊，若是能正常返回則關閉開關，不然繼續保持開關打開直到下次timeout。這樣業務恢復以後就能正常服務請求。

另外，爲了防止單個調用方的非法調用對服務的影響，服務也支持了多個維度限流，包括調用方AppId/ip限流和服務限流，接口限流等。
 

4、性能&擴展

 

因爲在線旅遊行業近幾年的高速增加，攜程做爲行業領頭羊也蓬勃發展，所以訪問量和數據量也大幅提高。公司對業務的要求是能夠支撐10倍容量擴展，擴展最難的部分在數據層，由於涉及到存量數據的遷移。

實時用戶行爲系統的數據層包括Redis和MySQL，Redis由於實現了一致性哈希，擴容時只要加機器，並對分配到新分區的數據做讀補償就能夠。

MySQL方面，咱們也作了水平切分做爲擴展的準備，分片數量的選擇考慮爲2的n次方，這樣作在擴容時有明顯的好處。由於攜程的mysql數據庫如今廣泛採用的是一主一備的方式，在擴容時能夠直接把備機拉平成第二臺（組）主機。假設原來分了2個庫，d0和d1，都放在服務器s0上，s0同時有備機s1。擴容只須要以下幾步：

     1.確保s0 -> s1同步順利，沒有明顯延遲
     2.s0暫時關閉讀寫權限
     3.確認s1已經徹底同步s0更新
     4.s1開放讀寫權限
     5.d1的dns由s0切換到s1

     6.s0開放讀寫權限

 

遷移過程利用MySQL的複製分發特性，避免了繁瑣易錯的人工同步過程，大大下降了遷移成本和時間。整個操做過程能夠在幾分鐘完成，結合DB降級的功能，只有在DNS切換的幾秒鐘時間會產生異常。

整個過程比較簡單方便，下降了運維負擔，必定程度也能下降過多操做形成相似GitLab式悲劇的可能性。
 

5、部署

 

前文提到Storm部署是比較方便的，只要上傳重啓就能夠完成部署。部署以後因爲程序從新啓動上下文丟失，能夠經過Kafka記錄的遊標找到以前處理位置，恢復處理。  另外有部分狀況下程序可能須要多版本運行，好比行爲紀錄暫時有多個版本，這種狀況下咱們會新增一個backupJob，在backupJob中運行歷史版本。