驚！使用300行代碼創(chuàng)建一個分布式系統(tǒng)
構(gòu)建一個分布式系統(tǒng)是很困難的。它需要可擴展性、容錯性、高可用性、一致性、可伸縮以及高效。為了達(dá)到這些目的，分布式系統(tǒng)需要很多復(fù)雜的組件以一 種復(fù)雜的方式協(xié)同工作。例如，Apache Hadoop在大型集群上并行處理TB級別的數(shù)據(jù)集時，需要依賴有著高容錯的文件系統(tǒng)（HDFS）來達(dá)到高吞 吐量。
在之前，每一個新的分布式系統(tǒng)，例如Hadoop和Cassandra，都需要構(gòu)建自己的底層架構(gòu)，包括消息處理、存儲、網(wǎng)絡(luò)、容錯性和可伸縮性。 慶幸的是，像Apache Mesos這樣的系統(tǒng)，通過給分布式系統(tǒng)的關(guān)鍵構(gòu)建模塊提供類似操作系統(tǒng)的管理服務(wù)，簡化了構(gòu)建和管理分布式系統(tǒng)的任務(wù)。 Mesos抽離了CPU、存儲和其它計算資源，因此開發(fā)者開發(fā)分布式應(yīng)用程序時能夠?qū)⒄麄€數(shù)據(jù)中心集群當(dāng)做一臺巨型機對待。
構(gòu)建在Mesos上的應(yīng)用程序被稱為框架，它們能解決很多問題：Apache Spark，一種流行的集群式數(shù)據(jù)分析工具；Chronos，一個類 似cron的具有容錯性的分布式scheduler，這是兩個構(gòu)建在Mesos上的框架的例子。構(gòu)建框架可以使用多種語言，包括 C++，Go，Python，Java，Haskell和 Scala。
在分布式系統(tǒng)用例上，比特幣開采就是一個很好的例子。比特幣將為生成 acceptable hash 的挑戰(zhàn)轉(zhuǎn)為驗證一塊事務(wù)的可靠性。可能需要 幾十年，單臺筆記本電腦挖一塊可能需要花費超過150年。結(jié)果是，有許多的“采礦池”允許采礦者將他們的計算資源聯(lián)合起來以加快挖礦速度。 Mesosphere的一個實習(xí)生，Derek，寫了一個比特幣開采框架，利用集群資源的優(yōu)勢來做同樣的事情。在接下來的內(nèi)容中，會以他的代碼為例。
1個Mesos框架有1個scheduler 和1個executor組成。scheduler 和Mesos master通信并決定運行什么任 務(wù)，而executor 運行在slaves上面，執(zhí)行實際任務(wù)。大多數(shù)的框架實現(xiàn)了自己的scheduler，并使用1個由Mesos提供的標(biāo)準(zhǔn) executors。當(dāng)然，框架也可以自己定制executor。在這個例子中即會編寫定制的scheduler，并使用標(biāo)準(zhǔn)命令執(zhí)行器 （executor）運行包含我們比特幣服務(wù)的Docker鏡像。
對這里的scheduler來說，需要運行的有兩種任務(wù)—— 單礦服務(wù)器任務(wù)和多礦服務(wù)器任務(wù)。服務(wù)器會和一個比特幣采礦池通信，并給每個“工人”分配塊。“工人”會努力工作，即開采比特幣。
任務(wù)實際上被封裝在executor框架中，因此任務(wù)運行意味著告訴Mesos master在其中一個slave上面啟動一個executor。 由于這里使用的是標(biāo)準(zhǔn)命令執(zhí)行器（executor），因此可以指定任務(wù)是二進(jìn)制可執(zhí)行文件、bash腳本或者其他命令。由于Mesos支持 Docker，因此在本例中將使用可執(zhí)行的Docker鏡像。Docker是這樣一種技術(shù)，它允許你將應(yīng)用程序和它運行時需要的依賴一起打包。
為了在Mesos中使用Docker鏡像，這里需要在Docker registry中注冊它們的名稱：
- const (
- MinerServerDockerImage = "derekchiang/p2pool"
- MinerDaemonDockerImage = "derekchiang/cpuminer"
- )
然后定義一個常量，指定每個任務(wù)所需資源：
- const (
- MemPerDaemonTask = 128 // mining shouldn't be memory-intensive
- MemPerServerTask = 256
- CPUPerServerTask = 1 // a miner server does not use much CPU
- )
現(xiàn)在定義一個真正的scheduler，對其跟蹤，并確保其正確運行需要的狀態(tài)：
- type MinerScheduler struct {
- // bitcoind RPC credentials
- bitcoindAddr string
- rpcUser string
- rpcPass string
- // mutable state
- minerServerRunning bool
- minerServerHostname string
- minerServerPort int // the port that miner daemons
- // connect to
- // unique task ids
- tasksLaunched int
- currentDaemonTaskIDs []*mesos.TaskID
- }
這個scheduler必須實現(xiàn)下面的接口：
- type Scheduler interface {
- Registered(SchedulerDriver, *mesos.FrameworkID, *mesos.MasterInfo)
- Reregistered(SchedulerDriver, *mesos.MasterInfo)
- Disconnected(SchedulerDriver)
- ResourceOffers(SchedulerDriver, []*mesos.Offer)
- OfferRescinded(SchedulerDriver, *mesos.OfferID)
- StatusUpdate(SchedulerDriver, *mesos.TaskStatus)
- FrameworkMessage(SchedulerDriver, *mesos.ExecutorID,
- *mesos.SlaveID, string)
- SlaveLost(SchedulerDriver, *mesos.SlaveID)
- ExecutorLost(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID,
- int)
- Error(SchedulerDriver, string)
- }
現(xiàn)在一起看一個回調(diào)函數(shù)：
- func (s *MinerScheduler) Registered(_ sched.SchedulerDriver,
- frameworkId *mesos.FrameworkID, masterInfo *mesos.MasterInfo) {
- log.Infoln("Framework registered with Master ", masterInfo)
- }
- func (s *MinerScheduler) Reregistered(_ sched.SchedulerDriver,
- masterInfo *mesos.MasterInfo) {
- log.Infoln("Framework Re-Registered with Master ", masterInfo)
- }
- func (s *MinerScheduler) Disconnected(sched.SchedulerDriver) {
- log.Infoln("Framework disconnected with Master")
- }
Registered在scheduler 成功向Mesos master注冊之后被調(diào)用。
Reregistered在scheduler 與Mesos master斷開連接并且再次注冊時被調(diào)用，例如，在master重啟的時候。
Disconnected在scheduler 與Mesos master斷開連接時被調(diào)用。這個在master掛了的時候會發(fā)生。
目前為止，這里僅僅在回調(diào)函數(shù)中打印了日志信息，因為對于一個像這樣的簡單框架，大多數(shù)回調(diào)函數(shù)可以空在那里。然而，下一個回調(diào)函數(shù)就是每一個框架的核心，必須要認(rèn)真的編寫。
ResourceOffers在scheduler 從master那里得到一個offer的時候被調(diào)用。每一個offer包含一個集群上可以給框架使用的資源列表。資源通常包括CPU、內(nèi)存、端口和磁盤。一個框架可以使用它提供的一些資源、所有資源或者一點資源都不給用。
針對每一個offer，現(xiàn)在期望聚集所有的提供的資源并決定是否需要發(fā)布一個新的server任務(wù)或者一個新的worker任務(wù)。這里可以向每個 offer發(fā)送盡可能多的任務(wù)以測試***容量，但是由于開采比特幣是依賴CPU的，所以這里每個offer運行一個開采者任務(wù)并使用所有可用的CPU資 源。
- for i, offer := range offers {
- // … Gather resource being offered and do setup
- if !s.minerServerRunning && mems >= MemPerServerTask &&
- cpus >= CPUPerServerTask && ports >= 2 {
- // … Launch a server task since no server is running and we
- // have resources to launch it.
- } else if s.minerServerRunning && mems >= MemPerDaemonTask {
- // … Launch a miner since a server is running and we have mem
- // to launch one.
- }
- }
針對每個任務(wù)都需要創(chuàng)建一個對應(yīng)的TaskInfo message ，它包含了運行這個任務(wù)需要的信息。
- s.tasksLaunched++
- taskID = &mesos.TaskID {
- Value: proto.String("miner-server-" +
- strconv.Itoa(s.tasksLaunched)),
- }
- Task IDs由框架決定，并且每個框架必須是唯一的。
- containerType := mesos.ContainerInfo_DOCKER
- task = &mesos.TaskInfo {
- Name: proto.String("task-" + taskID.GetValue()),
- TaskId: taskID,
- SlaveId: offer.SlaveId,
- Container: &mesos.ContainerInfo {
- Type: &containerType,
- Docker: &mesos.ContainerInfo_DockerInfo {
- Image: proto.String(MinerServerDockerImage),
- },
- },
- Command: &mesos.CommandInfo {
- Shell: proto.Bool(false),
- Arguments: []string {
- // these arguments will be passed to run_p2pool.py
- "--bitcoind-address", s.bitcoindAddr,
- "--p2pool-port", strconv.Itoa(int(p2poolPort)),
- "-w", strconv.Itoa(int(workerPort)),
- s.rpcUser, s.rpcPass,
- },
- },
- Resources: []*mesos.Resource {
- util.NewScalarResource("cpus", CPUPerServerTask),
- util.NewScalarResource("mem", MemPerServerTask),
- },
- }
TaskInfo message指定了一些關(guān)于任務(wù)的重要元數(shù)據(jù)信息，它允許Mesos節(jié)點運行Docker容器，特別會指定name、task ID、container information以及一些需要給容器傳遞的參數(shù)。這里也會指定任務(wù)需要的資源。
現(xiàn)在TaskInfo已經(jīng)被構(gòu)建好，因此任務(wù)可以這樣運行：
driver.LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, &mesos.Filters{RefuseSeconds: proto.Float64(1)})
在框架中，需要處理的***一件事情是當(dāng)開采者server關(guān)閉時會發(fā)生什么。這里可以利用StatusUpdate 函數(shù)來處理。
在一個任務(wù)的生命周期中，針對不同的階段有不同類型的狀態(tài)更新。對這個框架來說，想要確保的是如果開采者server由于某種原因失敗，系統(tǒng)會Kill所有開采者worker以避免浪費資源。這里是相關(guān)的代碼：
- if strings.Contains(status.GetTaskId().GetValue(), "server") &&
- (status.GetState() == mesos.TaskState_TASK_LOST ||
- status.GetState() == mesos.TaskState_TASK_KILLED ||
- status.GetState() == mesos.TaskState_TASK_FINISHED ||
- status.GetState() == mesos.TaskState_TASK_ERROR ||
- status.GetState() == mesos.TaskState_TASK_FAILED) {
- s.minerServerRunning = false
- // kill all tasks
- for _, taskID := range s.currentDaemonTaskIDs {
- _, err := driver.KillTask(taskID)
- if err != nil {
- log.Errorf("Failed to kill task %s", taskID)
- }
- }
- s.currentDaemonTaskIDs = make([]*mesos.TaskID, 0)
- }
萬事大吉！通過努力，這里在Apache Mesos上建立一個正常工作的分布式比特幣開采框架，它只用了大約300行GO代碼。這證明了使用Mesos 框架的API編寫分布式系統(tǒng)是多么快速和簡單。
