作者介紹：

孫健波，InfoQ 專欄作者，碩士畢業(yè)于浙江大學(xué)計(jì)算機(jī)學(xué)院，2015 年底加入七牛，參與并負(fù)責(zé)七牛大數(shù)據(jù)平臺整條鏈路多模塊的架構(gòu)設(shè)計(jì)以及開發(fā)工作，包括大數(shù)據(jù)計(jì)算引擎、時(shí)序數(shù)據(jù)庫服務(wù)以及日志檢索服務(wù)。在此之前主要研究和參與 Cloudfoundry、Kubernetes、Docker 等容器與容器云相關(guān)工作，是《Docker 容器與容器云》一書的主要作者之一。

大數(shù)據(jù)

圖 1

如圖 1 可以看到，現(xiàn)在大數(shù)據(jù)的生態(tài)相對來說比較成熟了，有很多很多大數(shù)據(jù)相關(guān)的組件。比如儲(chǔ)存 Kafka，HDFS，集群調(diào)度可以用 Mesos，檢索用 Spark 等。數(shù)據(jù)可視化還有Zeppelin 等等工具，監(jiān)控可以用 Grafan 等等。但是這么多復(fù)雜的組件放在一起，如果是玩技術(shù)可能玩的很嗨，有這么多組件可以玩，每個(gè)東西我們可以組合起來，可以完成自己的事情。但是對于一個(gè)只想要做業(yè)務(wù)，只關(guān)心業(yè)務(wù)，需要挖掘他們價(jià)值的人來說，這個(gè)其實(shí)非常痛苦的，因?yàn)樗麄冃枰涝鯓硬拍馨堰@些組件聯(lián)合起來，去把一個(gè)個(gè)組件的坑填上去。

Pandora 大數(shù)據(jù)平臺

圖 2

七牛大數(shù)據(jù)平臺是一個(gè)讓大家能夠更容易挖掘數(shù)據(jù)價(jià)值、讓大家更容易去使用大數(shù)據(jù)服務(wù)的平臺。

所以我們潘多拉七牛大數(shù)據(jù)團(tuán)隊(duì)做了以下的事情：

• 將多樣的大數(shù)據(jù)工具整合

• 將復(fù)雜的大數(shù)據(jù)管理簡化

• 構(gòu)建完整的大數(shù)據(jù)生命周期閉環(huán)（收集－加工－分析－管理－消費(fèi)－冷凍）

簡單來說（如圖 2），就是把各種各樣的組件根據(jù)你需要，像圖 2 這樣的一個(gè)操作界面圖一樣，每個(gè)組件都是一個(gè)很小的模塊單元。你可以根據(jù)自己的需要做一些計(jì)算，實(shí)時(shí)的、離線的。然后保存到相應(yīng)的位置上，如果需要儲(chǔ)存，我們有七牛云存儲(chǔ)，廉價(jià)的，穩(wěn)定的，可靠的服務(wù)。如果需要檢索你的日志，做搜索引擎相關(guān)的事情，就可以做日志檢索服務(wù)。如果需要做監(jiān)控相關(guān)等等實(shí)時(shí)的非常高效的查詢，就可以導(dǎo)出到我們的云數(shù)據(jù)庫服務(wù)。此外還有一些支持開源的的工具，大家可以非常輕松簡單、沒有使用門檻使用這個(gè)大數(shù)據(jù)平臺。

所以簡單的說 Pandora 大數(shù)據(jù)平臺的理念是什么呢？就是把各種各樣的大數(shù)據(jù)的工具整合起來，讓大家這個(gè)操作簡化，可以關(guān)注數(shù)據(jù)本身的價(jià)值，完成這個(gè)數(shù)據(jù)信息的挖掘。

Pandora 架構(gòu)

圖 3

圖 3 Pandora 的產(chǎn)品架構(gòu)，Pandora 的架構(gòu)很簡單，首先用戶會(huì)關(guān)心數(shù)據(jù)在本地，還是在其他的地方，這樣才能知道數(shù)據(jù)怎樣才能最簡單的放到系統(tǒng)上。我們考慮到有這樣的需求，所以做了一個(gè)工具叫 logkit，它可以直接從數(shù)據(jù)源導(dǎo)數(shù)據(jù)。如果你是我們七牛云儲(chǔ)存的用戶，什么都不要做就可以使用我們這個(gè)系統(tǒng)。然后根據(jù)你不同的需要，比如有一些客戶，他們要去爬蟲，要抓一大堆數(shù)據(jù)，可能他們就要清洗他們的網(wǎng)頁，我們也有為他們導(dǎo)出爬蟲或者導(dǎo)出數(shù)據(jù)以后，比如他們的原數(shù)據(jù)非常非常的多，每秒上千上萬上億的數(shù)據(jù)，這樣的話客戶不直接關(guān)心這些原始的數(shù)據(jù)，客戶會(huì)做一些篩選，做一些聚合。然后根據(jù)需要，導(dǎo)入到 Pandora 保存到日志檢索服務(wù)，或者查詢，快速檢索。

我們不僅有開放 API 用戶可以直接調(diào)用，在我們這個(gè)平臺上，為你的用戶提供你自己數(shù)據(jù)的價(jià)值，同時(shí)也可以基于開源的工具來查看你的這些數(shù)據(jù)。當(dāng)然最終如果你針對這些比較久的數(shù)據(jù)，你還想做離線的分析，那么你云儲(chǔ)存上面的數(shù)據(jù)，或者日志檢索服務(wù)上的數(shù)據(jù)，都可以通過我們這個(gè) Xspark 工具做分析，通過展示的工具來展示，還可以保存到你其他自己的服務(wù)里面。    

圖 4

圖 4 是 Pandora 的系統(tǒng)架構(gòu)，最外層的是數(shù)據(jù)源，我們把數(shù)據(jù)拿過來，進(jìn)入消息隊(duì)列，然后進(jìn)行轉(zhuǎn)換進(jìn)行自定義算。然后我們有一個(gè)導(dǎo)出服務(wù)，可以導(dǎo)出實(shí)時(shí)數(shù)據(jù)庫，還可以導(dǎo)出到日志檢索服務(wù)。然后進(jìn)行一個(gè) API，或者開源的實(shí)施數(shù)據(jù)可視化的工具，最后如果還要離線計(jì)算的話可以用 Xspark 進(jìn)行數(shù)據(jù)的分析和計(jì)算。   

 圖 5

Pandora 用了哪些組件呢？最上層是一個(gè)我們提供的代理，這樣的話，可以把你的數(shù)據(jù)從數(shù)據(jù)源里面導(dǎo)出，然后進(jìn)入接收數(shù)據(jù)的服務(wù)，再進(jìn)入消息隊(duì)列，這個(gè)是我們定制化的一個(gè) Kafka，下一步進(jìn)入數(shù)據(jù)轉(zhuǎn)換（過濾、清洗、計(jì)算等等），然后經(jīng)過定制的 Spark  Streaming。最后提供一個(gè)導(dǎo)出服務(wù)，導(dǎo)出服務(wù)可以導(dǎo)出到你其他想要的服務(wù)，比如說時(shí)序數(shù)據(jù)庫，這個(gè)是我們自研的分布式時(shí)序數(shù)據(jù)庫，用于實(shí)時(shí)的數(shù)據(jù)監(jiān)控、聚合等需求。

也可以導(dǎo)出到日志檢索服務(wù)，用于數(shù)據(jù)查詢分析，然后還可以導(dǎo)出到七牛的云儲(chǔ)存，最后經(jīng)過 Grafana、Kibana 等工具進(jìn)行數(shù)據(jù)可視化。無論是離線還是實(shí)時(shí)，均可以這樣處理。

圖 6

目前七牛的大數(shù)據(jù)團(tuán)隊(duì)有比較高的數(shù)據(jù)規(guī)模，每天上百TB，2000多億條實(shí)時(shí)的增量數(shù)據(jù)。我們提供的下游的的落地工具也是比較豐富的?；緷M足了目前我們看到的一些大數(shù)據(jù)方面的使用的需求。

那么這種量級的數(shù)據(jù)導(dǎo)出，到底會(huì)有哪些問題？

大家可以看到，我們有很多服務(wù)，像實(shí)時(shí)數(shù)據(jù)庫、日志檢索、云儲(chǔ)存等等，我們要把這些海量的數(shù)據(jù)經(jīng)過一些計(jì)算，然后再導(dǎo)出。這里面就是海量數(shù)據(jù)會(huì)在我們的系統(tǒng)里面經(jīng)過數(shù)次的變化，然后流動(dòng)的效率怎么辦呢？會(huì)不會(huì)有什么問題呢？大家馬上就想到，最大的問題就是延遲。我們號稱實(shí)時(shí)，如果有很大的延遲的話，用戶肯定沒有辦法接受的，這樣就沒有意義了。所以我們做了很大的工作，就是怎么樣把這個(gè)延遲降下來。

圖 7

在解決數(shù)據(jù)的延遲之前先來看一下數(shù)據(jù)傳輸?shù)哪Ｐ褪窃趺礃拥?。傳統(tǒng)的數(shù)據(jù)傳輸過程有兩種，就跟你收快遞是一樣的，比如你可以讓快遞放在那里你自己去拿，也可以讓他直接送到你手上。所以這是兩種模式，一種是你去用戶那邊拿數(shù)據(jù)，就是拉的模型。還有一種就是讓用戶直接把這個(gè)數(shù)據(jù)打過來。顯然作為這樣的一個(gè)服務(wù)，你只能讓用戶打過來，對用戶承諾，這個(gè)數(shù)據(jù)打到你這邊來他就可以放心了。

但是接受用戶打過來的數(shù)據(jù)有什么問題呢？就是說你這個(gè)數(shù)據(jù)的效率，其實(shí)取決于用戶客戶打過來的姿勢。如果對應(yīng)不同的下游服務(wù)的話，可能用戶使用的姿勢不同，如果寫程序的話，那么你只是普通的連一下，然后導(dǎo)一下，你就會(huì)遇到很多的問題。所以姿勢非常的重要。

所以我們可以看一下，數(shù)據(jù)傳輸有幾種，一般構(gòu)想的，可以保證的東西，常見的東西，你會(huì)覺得數(shù)據(jù)的導(dǎo)出，流量是不會(huì)有太多變化的，比如說一個(gè)用戶今天是 10 MB/s，那么他明天會(huì)不會(huì)變成 100 MB/s呢？最后大家想的是 20 MB/s等等，不會(huì)想到 100 MB/s。這個(gè)變化一定嗎？我覺得未必。尤其像我們作為一個(gè) PaaS 的廠商，就沒辦法去說，一定要想用戶今天是 10 MB/s，明天有可能是還 15，20 MB/s，但是我們要時(shí)刻準(zhǔn)備著他是 100 MB/s打過來。

然后我們還可能會(huì)想數(shù)據(jù)的下游服務(wù)是穩(wěn)定可靠的話，我們提供的相當(dāng)于一個(gè)數(shù)據(jù)的變化，大數(shù)據(jù)的分析。我們提供了非常多的下游的服務(wù)，那么很多人覺得下游是非常穩(wěn)定的。但是下游的這個(gè)可靠性其實(shí)是不太確定的。像之前也爆出很多的廠商，知名國外的廠商也會(huì)有這種問題。所以你很難保證數(shù)據(jù)的下游一定是穩(wěn)定的。

數(shù)據(jù)傳輸常見的情況：

1. 導(dǎo)出的上游數(shù)據(jù)產(chǎn)量是穩(wěn)定不變(變化緩慢)的 

2. 導(dǎo)出的下游服務(wù)永遠(yuǎn)是穩(wěn)定可用（鏈路損耗嚴(yán)重） 

3. 導(dǎo)出的速度僅受限于上下游中的一方影響

• 單向吞吐量= 請求大小*并發(fā)數(shù)

• 整體吞吐量= f(拉取吞吐量,鏈路承載能力,推送吞吐量)

• 舉例 :流量 20k/s = 上游 10k/s*2+下游 5k/s*2 ?網(wǎng)絡(luò)抽風(fēng)?下游響應(yīng)慢? 網(wǎng)卡打滿?內(nèi)存超限?

接著大家可能最容易想到的就是數(shù)據(jù)的導(dǎo)出或者傳輸?shù)乃俣?，就是上游、下游的速度下線，再取最小，實(shí)際上真的是這樣的嗎？實(shí)際上是一般認(rèn)為并發(fā)數(shù)乘以每一個(gè)請求的大小，就是實(shí)際的總量。那我們的整體請求是怎樣的呢？就是上游你拉數(shù)據(jù)，然后去下游打數(shù)據(jù)的量，最后你忽略了一個(gè)過程，就是這個(gè)傳輸鏈路的承載能力的推送吞吐量。舉例來說，如果我們的流量是是 20 K/s 的話，那我們上游的請求是 10K×2，兩個(gè)并發(fā)，下游是 5K×4。這樣真的可以嗎？未必可以。因?yàn)槲覀儠?huì)遇到像網(wǎng)絡(luò)不穩(wěn)定，下游相應(yīng)慢，內(nèi)存超限等等這些問題，所以其實(shí)這都是我們必須要考慮到的。

那么怎樣去解決這個(gè)問題呢？我們可以想到一些比較常見的思路。

• 上下游解耦：拉取與推送解耦，數(shù)據(jù)預(yù)取、隊(duì)列暫存、拉取與發(fā)送并行

• 任務(wù)分割：大任務(wù)分解成小任務(wù)，小任務(wù)水平擴(kuò)展

• 任務(wù)標(biāo)準(zhǔn)化：每個(gè)任務(wù)承載固定的流量，流量增加則增加任務(wù)數(shù)量

• 提升資源利用率：調(diào)度、平衡、壓榨機(jī)器性能

• 提供任務(wù)管理能力：運(yùn)維、運(yùn)營、監(jiān)控

• 更懂下游

上下游的解耦是怎樣的呢？就是拉取數(shù)據(jù)與推送解耦分開來，中間提供一個(gè)隊(duì)列，這樣可以暫存數(shù)據(jù)，這樣就被認(rèn)為這個(gè)數(shù)據(jù)的速度，其實(shí)是相對來說快的。你只要保證解歐的隊(duì)列不出問題就可以了。

還會(huì)想到什么呢？如果一個(gè)用戶今天 10 MB/s，明天變成 100 MB/s了，這樣你原來的服務(wù)肯定扛不住的。你要把這個(gè) 100 MB/s變成十個(gè) 10MB/s，那么這個(gè)問題就可以輕松的搞定了。再者就是任務(wù)標(biāo)準(zhǔn)化，我們經(jīng)常會(huì)提的服務(wù)混部、這里用一個(gè) 5 核的機(jī)器、那里用一個(gè) 10 核的機(jī)器，實(shí)際上這樣對你的服務(wù)影響非常大。如果你能把你的數(shù)據(jù)，標(biāo)準(zhǔn)化起來，同時(shí)我們所有的集群都是用同樣規(guī)模的機(jī)器，那么在你做這個(gè)策略的時(shí)候就可以簡化很多思考，同時(shí)你可以保證，你的這個(gè)任務(wù)分割，就是把大任務(wù)化解成小任務(wù)這個(gè)事情是可靠的。然后我們還想到怎么樣提升這個(gè)利用率，管理能力調(diào)度，監(jiān)控運(yùn)營等等，最主要我們要更懂下游服務(wù)，比如實(shí)時(shí)數(shù)據(jù)庫，日志檢索等等，我們最終的目標(biāo)是把這個(gè)延遲降下來。

構(gòu)建 Pandora 加速系統(tǒng)

圖 8

剛剛已經(jīng)我們看到有導(dǎo)出服務(wù)，導(dǎo)出服務(wù)就是我們的 Xspark，Xspark 已經(jīng)做了很多很多任務(wù)，最重要的是它很輕。它做了哪些呢？就是剛剛看到的數(shù)據(jù)導(dǎo)出，從 Xspark 里面導(dǎo)出，還有數(shù)據(jù)的過濾轉(zhuǎn)換、精細(xì)化的調(diào)度等等。精細(xì)化是什么概念呢？你不光考慮 CPU，考慮內(nèi)存，同時(shí)還要看出網(wǎng)卡，機(jī)器的規(guī)模等等一系列的考慮。所以它做的事情非常多，最主要是這個(gè)精細(xì)化的調(diào)度。然后我們構(gòu)建了一個(gè)輕量級的分布式的 goroutine 來做這個(gè)事情的。它可以提供一個(gè)非常強(qiáng)的保證，保證我們的導(dǎo)出服務(wù)，如果下游出了問題，完全不會(huì)影響其他的服務(wù)。但是我們今天服務(wù)的重點(diǎn)不是在講這個(gè)導(dǎo)出，而是講我們要講，我們怎樣構(gòu)建一套更好的加速器，來加速導(dǎo)出服務(wù)。

－加速系統(tǒng)的選型

• logstash？

• beat？

• flume？

• 自研？

那么在構(gòu)建加速系統(tǒng)的時(shí)候你就會(huì)想選型的問題。一開始我們最先遇到的是我們的日志檢索服務(wù)，對應(yīng)其中的一個(gè)插件怎么處理，常規(guī)的借用社區(qū)的解決方法，像 logstash、beat、flume 等等的工具。那我們調(diào)研下來什么概念呢？就是對應(yīng)我們剛才說的這些思考，比如上下游解耦等等。我們發(fā)現(xiàn)像 logstash 它更多注重做客戶端搜集的事情，它作為一個(gè)中間端，或者服務(wù)端它接觸數(shù)據(jù)然后再打向各個(gè)服務(wù)，其實(shí)它做的并不好。而 beat 就是提供一個(gè)輕量級的搜集系統(tǒng)的工具。

圖 9

flume 提供這樣一個(gè)緩存（圖 9），我們覺得 flume 比較可靠，就去嘗試了一下，最大的問題就是它在不同的位置的情況下，如果你配置這么多用戶就可以了。但是如果你是一個(gè) PaaS 廠商，你提供的用戶是十萬個(gè)，你難道就配置十萬個(gè)讓它自動(dòng)生成嗎？這個(gè)實(shí)在太不優(yōu)雅了，也不符合我們 Gopher 的體會(huì)，所以我們就去自研了。

－語言的選擇

圖 10

我發(fā)現(xiàn)幾乎所有來大會(huì)上分享的老師都要回答一個(gè)話題，那就是為什么要選 Go ？其實(shí)用 Go 來做這個(gè)事情是很自然的選擇，不止再這一個(gè)模板，在大數(shù)據(jù)里面做了很多很多的組件也都是使用 Go 來寫的。那么我們來對比下我們的需求，從需求出發(fā)看語言的選型。

首先上下游解耦怎么做呢？這個(gè)就是有一個(gè) buffer 的概念，是不是可以把數(shù)據(jù)有一個(gè)接收，然后傳到 buffer  channel 里面，然后另外一端從 buffer  channel 里面拿數(shù)據(jù)。之后任何一個(gè)水平擴(kuò)展怎么辦呢？肯定會(huì)想到水平擴(kuò)展就是分布式了。那分布式怎么處理呢？一般都是進(jìn)程級的，那協(xié)程級的呢？協(xié)程級的會(huì)不會(huì)更舒服呢？因?yàn)橐呀?jīng)有語言幫你做這個(gè)調(diào)度。然后你要提升這個(gè)資源利用率，提升任務(wù)管理能力，你是不是就可以把這個(gè)注意力專注到任務(wù)資源分配的調(diào)度管理等等方面。

然后最重要的更懂下游怎么辦呢？因?yàn)槭亲匝?，可以讓下游寫組件的小伙伴自寫對應(yīng)的服務(wù)就可以了，我們就可以把這個(gè)過程通過插件寫進(jìn)去。例如你是做日志檢索的，那么你寫一個(gè)加速你日志檢索的傳輸插件。當(dāng)然還有很多很多理由，比如 go 所有人都會(huì)說簡單，易學(xué)易用。社區(qū)經(jīng)過這么幾年的發(fā)展，已經(jīng)非?；钴S了，還有它的部署迭代更簡便。大家都知道 golang 編譯出來就是個(gè)二進(jìn)制的包，你怎么玩都行。然后它效率非常高，它很穩(wěn)定性能也高，并發(fā)編程，還有我們七牛的技術(shù)棧，基本就是 golang，所以我們堅(jiān)定不移的選擇了 golang。  

－核心模型

圖 11

然后我們看一下我們要做這個(gè)事情，如果我們要自己開發(fā)這個(gè)東西它的核心模型是怎樣的？首先你會(huì)想到你面對的是一個(gè)數(shù)據(jù)源。然后你要用事務(wù)的形式把數(shù)據(jù)接收進(jìn)來，為什么用事務(wù)的形式呢？我們后面再講。我們經(jīng)過一個(gè)隊(duì)列，很多人講這個(gè)隊(duì)列怎么做比較好。如果你真的是加速考慮的話，只有一個(gè)選擇，那就是內(nèi)存，否則的話，其他的性能都會(huì)遇到很大的瓶頸。然后下游 sink 可以自己寫各種各樣的插件，你想導(dǎo)出到什么服務(wù)就導(dǎo)出到什么服務(wù)。

關(guān)于 sink ，用插件形式的下游適配器的形式，因?yàn)闆]有人比下游更懂下游。就像我們老大陳超經(jīng)常說，情人節(jié)給你女朋友或者老婆送個(gè)紅包就行了，讓她自己買是最好的。就是這個(gè)道理，你把一個(gè)球給他扔過去，別人能不能承受得住這個(gè)重量，這個(gè)不好說，還是讓他自己來吧。

提到我們剛才說的，用事務(wù)的的形式做。就是如果你這個(gè)不行了，你跟你的導(dǎo)出服務(wù)說慢點(diǎn)導(dǎo)或者導(dǎo)出到別的服務(wù)。然后如果你行就直接放進(jìn)去。然后同時(shí)事務(wù)也是解決分布式的問題，我們本身在調(diào)度的過程中可以開多任務(wù)，那么怎么樣保證這個(gè)數(shù)據(jù)只流向一處，其實(shí)也是事務(wù)。事務(wù)可能大家可能考慮到一個(gè)問題，如果有一個(gè)鎖怎么辦？如果累計(jì)的數(shù)據(jù)在內(nèi)存里面?zhèn)鬟f的話，它只是把數(shù)據(jù)放到這個(gè) channel 里面，實(shí)際上這個(gè)數(shù)據(jù)傳輸非?？斓模@個(gè)鎖是非常小的。同時(shí)爭搶這個(gè)鎖的這些并發(fā)，如果你控制的好的話，實(shí)際上只有十幾個(gè)并發(fā)，或者幾十個(gè)并發(fā)在搶這個(gè)鎖的話，實(shí)際上這個(gè)鎖的性能非常低的。所以這個(gè)事務(wù)我們實(shí)踐過來以后用起來非常舒服的。

圖 12

此外還有一個(gè)問題，萬一需要重啟或者掛了怎么辦呢？對于重啟，需要提供一個(gè)策略，怎么樣讓這個(gè)內(nèi)存的隊(duì)列進(jìn)入到本地磁盤？我們用一個(gè) sink 把內(nèi)存里面的數(shù)據(jù)統(tǒng)統(tǒng)的進(jìn)入一個(gè)本地的磁盤隊(duì)列。然后根據(jù)你恢復(fù)的時(shí)候再把這個(gè)數(shù)據(jù)恢復(fù)過來，所以就解決的數(shù)據(jù)重啟的問題。

圖 13

如果掛了怎么辦？因?yàn)閽炝宋覀冞€有上游的導(dǎo)出服務(wù)，在這上我們可以認(rèn)為它是做了專注于自己的事情，可以給我們提供數(shù)據(jù)重播的能力。我們怎么辦呢？就是數(shù)據(jù)來了以后我們只要記錄最基本的元數(shù)據(jù)，如這套數(shù)據(jù)的 Offset 是從多少到多少，哪個(gè) patition，如果它成功發(fā)送到下游就 OK，這條數(shù)據(jù)就過去了。那么有一些數(shù)據(jù)它的 offset 從開始到結(jié)束放在這里一直沒有導(dǎo)出，那怎么辦呢？它故障了，我們就調(diào)用導(dǎo)出服務(wù)的重播能力，進(jìn)行數(shù)據(jù)的重播，這樣數(shù)據(jù)也不會(huì)丟失。

圖 14

然后我們可以看一下整體。我們整體的話，進(jìn)是通過事務(wù)來保證，出是通過事務(wù)來保證它到底有沒有 sink 出去。保證它多的協(xié)程同時(shí)去拿的話，不會(huì)實(shí)現(xiàn)沖突（保證數(shù)據(jù)只會(huì)有一份出來）。關(guān)停就走磁盤的隊(duì)列，啟動(dòng)再把這個(gè)磁盤的數(shù)據(jù)流回這個(gè) channel。然后用一個(gè)簡單的狀態(tài)機(jī)完成這個(gè)流程。

圖 15

在眾多模塊組合之后，整個(gè)框架基本上搭起來了，保證了任務(wù)的流動(dòng)。但是我們還要構(gòu)建一rest－api，讓別人數(shù)據(jù)能導(dǎo)過來，讓導(dǎo)出的數(shù)據(jù)能過來。 而且是需要構(gòu)建一個(gè)任務(wù)級的 rest－api，因?yàn)槊媾R的是 PaaS 上面百萬的用戶，肯定要把這個(gè)事情做成一個(gè)單一的某一個(gè)用戶的級別，那我們可以用 agent 來調(diào)度這個(gè)事情，封裝成 task 的概念來針對不同的用戶之后再導(dǎo)出到 sink 或者下游不同的服務(wù)。那么這樣一個(gè)單機(jī)版，看似可以分布式化的已經(jīng)完成了，而且相對比較簡單干凈。

－單機(jī)模型核心總結(jié)

• 一個(gè) repo 一個(gè)單獨(dú)的 task，不同 repo 間 task 不共享

• 一個(gè) task 包含一個(gè) MQ，多個(gè) sink、一個(gè)磁盤隊(duì)列

• 根據(jù) mongo 的配置決定 sink 的數(shù)量(與 capacity 相同)MQ 中包含 transaction 池，每次 sink，通過事務(wù)控制數(shù)據(jù)進(jìn)出 MQ 保證原子性

• 重啟的過程也以 sink 的形式，只不過發(fā)送端變?yōu)榇疟P，保證了服務(wù)升級時(shí)內(nèi)存數(shù)據(jù)不丟失

我們總結(jié)一下，最簡單的就是對一個(gè)用戶的數(shù)據(jù)對應(yīng)一個(gè) task，用戶單位是 repo，task 是不共享的，分別獨(dú)占了資源。一個(gè) task 包含一個(gè) MQ，多個(gè)sink，有一個(gè)磁盤隊(duì)列可以重啟等等。還可以從 mongo 里面拿到相應(yīng)的配置，使用 mongo。事務(wù)來控制原子性，重啟的過程也是以 sink 的形式，通過數(shù)據(jù)把下游打到磁盤，這樣整個(gè)事情就完成了。

分布式的困難

• 維護(hù)困難，如果我的集群要添加或者減少一個(gè) producer 節(jié)點(diǎn)，該怎么辦？

• 數(shù)據(jù)分散，所有的數(shù)據(jù)都有可能經(jīng)過任意 producer 節(jié)點(diǎn)，對應(yīng)每個(gè) producer 發(fā)往下游的數(shù)據(jù)分散，導(dǎo)致每個(gè)請求的可能聚集的數(shù)據(jù)量小，batchsize 小。

• 資源浪費(fèi)，每個(gè) producer 都要維護(hù)大量的 task，對應(yīng)大量的goroutine，浪費(fèi) CPU 和內(nèi)存。

• 負(fù)載不均，若是純粹的隨機(jī)，或者輪詢，一旦遇到機(jī)器配置不同或者服務(wù)混部等情況，負(fù)載無法均衡。

• 不易管理，新的業(yè)務(wù)需要啟動(dòng)對應(yīng)的 producer task，需要改變配置等，都無法操作。

注：producer 為加速服務(wù)名字

那么這樣單機(jī)版仿佛已經(jīng)解決問題了，還有什么問題呢？如果你簡單的把這個(gè)組件，我們的加速服務(wù)導(dǎo)出，放到很多機(jī)器上面是不是就解決問題了呢？

其實(shí)并不是，還有哪些問題呢？就是維護(hù)困難，如果我們的數(shù)據(jù)分散，怎么樣控制它在不同的機(jī)器上整合，或者平衡等等，資源浪費(fèi)怎么辦？剛剛也說了，我們有 task，如果有用戶過來創(chuàng)建一下或者試玩一下就再也不用了，怎么辦？怎么清理？還有負(fù)載不均衡怎么辦？還有管理起來怎么辦？

－分布式一致性問題

• zookeeper/etcd

• 自研分布式算法

• 最終一致性 =>  pull系統(tǒng)+版本戳

所以我們面臨著分布式服務(wù)里面一個(gè)常見的問題，就是怎么把數(shù)據(jù)傳遞到每一個(gè)節(jié)點(diǎn)，也就是分布式一致性的問題。簡單來說就是這個(gè)時(shí)候你怎么讓數(shù)據(jù)通知到每一個(gè)節(jié)點(diǎn)，讓一個(gè)節(jié)點(diǎn)都知道你做什么，你要解決什么樣的問題。其實(shí)現(xiàn)在社區(qū)發(fā)展這么多年，其實(shí)一致性問題解決起來也相對來說有比較成熟的方案?？赡軙?huì)選擇 etcd/zookeeper 去解決這個(gè)強(qiáng)一致的問題，還有一些自研的算法。

那么要不要自研呢？我們想了一下，如果我們做這樣一個(gè)加速服務(wù)，真的需要強(qiáng)一致嗎？如果一個(gè)數(shù)據(jù)過來我們導(dǎo)出，我們跟他說你導(dǎo)出到這里不太平衡，機(jī)器的負(fù)載不太好，你應(yīng)該用另外一個(gè)機(jī)器上使用另外一臺的機(jī)器的加速服務(wù)，這個(gè)消息的同步真的需要那么實(shí)時(shí)嗎？其實(shí)我們權(quán)衡下來并不是的，我們只要最終這個(gè)消息發(fā)送過來三五分鐘以后，能夠把這個(gè)事情達(dá)到一個(gè)非常協(xié)調(diào)的狀態(tài)，那么事情就解決了。他只要把這個(gè)數(shù)據(jù)讓我最終感知，達(dá)到最終的一致，這個(gè)事情就可以。所以我們就去拉這個(gè)源數(shù)據(jù)。加上版本戳保證數(shù)據(jù)的最終一致性。

圖 16

說到這個(gè)最終一致性，七牛自己有一套很好的二級緩存框架兩保證這個(gè)一致性。這個(gè)是怎么樣的呢？首先你的源數(shù)據(jù)肯定要一個(gè)數(shù)據(jù)庫做存儲(chǔ)。想要用的時(shí)候，如果源數(shù)據(jù)直接拉的話肯定要把它擊穿了。每一個(gè)請求過來都去訪問，幾乎所有的數(shù)據(jù)庫都扛不住這種壓力。這時(shí)候肯定會(huì)想到就是緩存。緩存是怎樣的呢？首先一個(gè)數(shù)據(jù)過來，同步到 mongo 的數(shù)據(jù)庫里面，然后做兩層緩存，一層就是本地的，去本地拿，發(fā)現(xiàn)本地沒有。那么再去二級緩存的服務(wù)器拿。然后拿了發(fā)現(xiàn)也沒有，這時(shí)候再去 mongo 里面拿。之后再把數(shù)據(jù)存在緩存、本地各一份，然后根據(jù)你的需要設(shè)置過期時(shí)間，這樣你的數(shù)據(jù)就緩沖的很好，相當(dāng)于你對這個(gè)數(shù)據(jù)庫本身的請求每兩分鐘才幾百次，上千次這樣，因?yàn)榇蟛糠值恼埱髷?shù)據(jù)已經(jīng)被緩存起來了。

這樣做還有什么好處？比如說數(shù)據(jù)庫掛了，我們還有二級緩存，這樣二級緩存掛了，我們還有本地緩存，這樣就保證了如果主服務(wù)也掛了，那么我其他的服務(wù)還可以繼續(xù)的工作。相當(dāng)我其實(shí)和 master 這個(gè)東西是解耦的，我不會(huì)受到 matser 掛掉的影響，如果數(shù)據(jù)有改變的話，已經(jīng)通知到其他的組件。  

圖 17

所以我們看一下我們最終要維護(hù)緩存里的指標(biāo)是什么？

1、保證狀態(tài)?？偛荒苷f啟動(dòng)了以后不能關(guān)，首先要保證這個(gè)啟動(dòng)能停止

2、要有分配的能力，自動(dòng)分配也好，手動(dòng)分配也好，要有一個(gè)分配的過程

3、要保證批量發(fā)送，有能力去調(diào)整發(fā)送發(fā)小

4、并發(fā)數(shù)，你要開多少加速的服務(wù)，每一個(gè)發(fā)到下游的請求有多少的并發(fā)數(shù)

5、隊(duì)列緩存容量有多大，超過多少后會(huì)反壓

6、消息的接受有多少的并發(fā)

7、如果要手動(dòng)指定機(jī)器的話，就可以指定一下

   圖 18

然后我們考慮到如果這個(gè)任務(wù)一會(huì)在這個(gè)機(jī)器，一會(huì)在那個(gè)機(jī)器的話，其實(shí)是對鏈路是浪費(fèi)的，首先網(wǎng)卡就非常浪費(fèi)，然后啟停等等，調(diào)度的過程是非常浪費(fèi)的。所以我們剛剛已經(jīng)說了，我們是基于任務(wù)的標(biāo)準(zhǔn)化，每個(gè)任務(wù)其實(shí)是固定大小的規(guī)模。所以如果每個(gè)任務(wù)都已經(jīng)是固定大小的規(guī)模，我們可以穩(wěn)定把它分配到某些機(jī)器上。

基于一個(gè)最簡單的，首先多少任務(wù)已經(jīng)知道了，排一個(gè)序，然后根據(jù)它需要的數(shù)量我們給它足夠的機(jī)器的分配?？赡苓€會(huì)遇到一些其他的問題，比如機(jī)器的配置不太均衡，當(dāng)然最好還是均衡一點(diǎn)，但是總有一些難免的情況。那么你可以通過手動(dòng)指定的情況把某一個(gè)任務(wù)綁定到指定的機(jī)器上面，然后大概的配置，調(diào)度算法就完成了。非常簡單，穩(wěn)定，我們用平衡的任務(wù)標(biāo)準(zhǔn)化的機(jī)制解決這個(gè)問題。

－白名單機(jī)器綁定

• 手動(dòng)與自動(dòng)相結(jié)合

• 應(yīng)對突發(fā)流量

• 防止大任務(wù)的抖動(dòng)

• 彌補(bǔ)機(jī)器配置帶來的差異化

• admin 能力

此外我們還會(huì)把這個(gè)機(jī)器手動(dòng)綁定的能力加上，這個(gè)綁定的能力大概是一個(gè)怎樣的概念呢？首先你可以手動(dòng)和自動(dòng)相結(jié)合，我們剛開始寫代碼的時(shí)候，盲目的相信自動(dòng)化的過程，覺得我寫一個(gè)厲害的算法什么都是自動(dòng)的，只要這個(gè)算法夠厲害就沒問題。但是真正線上的服務(wù)總歸有出人意料的事情發(fā)生，你肯定要加上手動(dòng)的能力。那手動(dòng)能力有什么好處呢？就是應(yīng)對突發(fā)流量。萬一真的來不及擴(kuò)容怎么辦？可以臨時(shí)調(diào)整，非常靈活，防止大任務(wù)的抖動(dòng)。

比如我們就很多的大客戶，他們的數(shù)據(jù)量非常巨大，我們可以給他指定一些集群和機(jī)器，把他們綁定在上面。這樣相對來說這個(gè)大客戶的數(shù)據(jù)是比較穩(wěn)定的，這個(gè)任務(wù)就不會(huì)抖動(dòng)，不會(huì)影響別的資源，不會(huì)侵蝕小客戶。所以小客戶和大客戶的體驗(yàn)都是非常好的。然后還有一個(gè)彌補(bǔ)機(jī)器配飾帶來的差異化，你可以有一些手動(dòng)化的機(jī)制。

我們還會(huì)提供一些 API，來做什么呢？就是獲取監(jiān)控信息－任務(wù)數(shù)量、成功失敗率、lag等等指標(biāo)，還有提供一些管理接口，看一下歷史的問題，然后我們看一下問題就可以通過這些來解決。

－按需創(chuàng)建與資源回收

• 數(shù)據(jù)到達(dá)時(shí)才創(chuàng)建，按需創(chuàng)建。

• producer 實(shí)例中某個(gè) task 如果長時(shí)間沒有數(shù)據(jù)過來，則銷毀掉對應(yīng)的 task，釋放資源。

上述提到如果有用戶來創(chuàng)建，創(chuàng)建了一會(huì)覺得不太好用怎么辦？就不用了，那不用這個(gè)資源肯定要浪費(fèi)了，那資源怎么回收呢？其實(shí)有很多的小客戶他們是試用的性質(zhì)，這時(shí)候我們提供一些免費(fèi)的額度，這樣他很快就用完了，很快用完了他們不想付費(fèi)了，而且他們也沒有真正的的有需要。那么他這個(gè)數(shù)據(jù)資源就會(huì)占在那邊，很多 task 應(yīng)該會(huì)碰到這樣的問題，就是資源回收的問題。那資源回收怎么解決呢？最簡單是基于對過去的統(tǒng)計(jì)進(jìn)行一個(gè)預(yù)測。比如現(xiàn)在一直是在打數(shù)據(jù)，他突然有一會(huì)沒有打數(shù)據(jù)了，可能是一天，也可能是多少小時(shí)，那么你界定一下，這個(gè)數(shù)據(jù)資源可以釋放掉，然后快速的起停。 

－protobuf 序列化協(xié)議

• 通過 protobuf 協(xié)議與上游通信

• 不重復(fù)解析數(shù)據(jù)，去除 json 等解析的 cpu 消耗

我們使用了序列化協(xié)議，在這塊會(huì)用到 protobuf，這個(gè)效果非常好。對比一下，如果你用 json 序列化協(xié)議， cpu 的消耗和 protobuf 大概有十倍差距。如果你能用 protobuf 的話，盡量用這個(gè)，這個(gè)帶來的體驗(yàn)是非常好的。

－變長的失敗等待時(shí)間

• 向下游寫入失敗，則休眠 1s 再重試，依然失敗則休眠時(shí)間增加，一直到 10s 為止

• 若寫入成功，則失敗的休眠時(shí)間重置為 1s

• 有效減少下游壓力

最后很重要的一點(diǎn)，變長失敗等待的時(shí)間。比如你這一次訪問出問題，那么下一次你再去訪問，下下游還是掛了。那我給你一秒，如果你還是掛了，那我就等再一秒，因?yàn)槲医o你一個(gè)等待的機(jī)會(huì)。因?yàn)槲覀兘?jīng)常發(fā)現(xiàn)像數(shù)據(jù)打服務(wù)打掛了，很多時(shí)候一直在打，會(huì)有數(shù)據(jù)堆積，成堆的數(shù)據(jù)打過來的話，這樣對下游會(huì)造成一個(gè)崩敗式的過程。所以我們給他等待的機(jī)會(huì)，給他休息一秒，再休息三秒，等到一個(gè)預(yù)值十秒，如果它恢復(fù)了，我們再回到正常的過程。這樣可以有效減少下游的壓力，讓下游快速的恢復(fù)，然后我們把這個(gè)數(shù)據(jù)快速的傳輸過去。

Pandora 加速系統(tǒng)的成果

• 無數(shù)據(jù)重復(fù)寫入、無數(shù)據(jù)丟失問題

• 寫入更平滑，去除毛刺

• 更高的機(jī)器資源利用率

• 更懂下游

• 沒有Lag！

所以最終這個(gè)加速服務(wù)構(gòu)建了哪些成果呢？首先最重要就是沒有數(shù)據(jù)重復(fù)的問題，也沒有數(shù)據(jù)丟失的問題?？梢宰寯?shù)據(jù)的寫入更加平滑，就是去除毛刺。這個(gè)毛刺是什么概念呢？就是玩過大數(shù)據(jù)的，或者有一定數(shù)據(jù)量的朋友都會(huì)感受到，有時(shí)候你不同的機(jī)器，或者不同的組件，不同的實(shí)例去打的時(shí)候，他們這個(gè)時(shí)間是不一樣的。因?yàn)楦鶕?jù)你的請求，這個(gè)可能十秒返回，那個(gè)可能一秒就返回了，這個(gè)就是毛刺，你會(huì)認(rèn)為所有的實(shí)例都返回的，你才認(rèn)為這個(gè)請求OK。那我們做的這個(gè)導(dǎo)出的加速服務(wù)器，就是把這個(gè)毛刺解決了，這樣整個(gè)上下游之間的數(shù)據(jù)傳輸?shù)男史浅Ｆ交恢痹谝粋€(gè)非常高的水平。同時(shí)我們提高的機(jī)器資源的利用率。也更懂下游，因?yàn)槲覀円圆寮哪Ｊ骄帉懻麄€(gè)下游的服務(wù)，最終達(dá)到的效果是沒有延遲。

本文來自：微信公眾平臺

感謝作者：孫健波

查看原文：Go 在萬億級大數(shù)據(jù)平臺開發(fā)中的實(shí)戰(zhàn)