構建一個分布式系統是很困難的。它需要可擴展性、容錯性、高可用性、一致性、可伸縮以及高效。為了達到這些目的，分布式系統需要很多復雜的組件以一 種復雜的方式協同工作。例如，Apache Hadoop在大型集群上并行處理TB級別的數據集時，需要依賴有著高容錯的文件系統（HDFS）來達到高吞 吐量。

在之前，每一個新的分布式系統，例如Hadoop和Cassandra，都需要構建自己的底層架構，包括消息處理、存儲、網絡、容錯性和可伸縮性。 慶幸的是，像Apache Mesos這樣的系統，通過給分布式系統的關鍵構建模塊提供類似操作系統的管理服務，簡化了構建和管理分布式系統的任務。 Mesos抽離了CPU、存儲和其它計算資源，因此開發者開發分布式應用程序時能夠將整個數據中心集群當做一臺巨型機對待。

[[143382]]

構建在Mesos上的應用程序被稱為框架，它們能解決很多問題：Apache Spark，一種流行的集群式數據分析工具；Chronos，一個類 似cron的具有容錯性的分布式scheduler，這是兩個構建在Mesos上的框架的例子。構建框架可以使用多種語言，包括 C++，Go，Python，Java，Haskell和 Scala。

在分布式系統用例上，比特幣開采就是一個很好的例子。比特幣將為生成 acceptable hash 的挑戰轉為驗證一塊事務的可靠性。可能需要 幾十年，單臺筆記本電腦挖一塊可能需要花費超過150年。結果是，有許多的“采礦池”允許采礦者將他們的計算資源聯合起來以加快挖礦速度。 Mesosphere的一個實習生，Derek，寫了一個比特幣開采框架，利用集群資源的優勢來做同樣的事情。在接下來的內容中，會以他的代碼為例。

1個Mesos框架有1個scheduler 和1個executor組成。scheduler 和Mesos master通信并決定運行什么任 務，而executor 運行在slaves上面，執行實際任務。大多數的框架實現了自己的scheduler，并使用1個由Mesos提供的標準 executors。當然，框架也可以自己定制executor。在這個例子中即會編寫定制的scheduler，并使用標準命令執行器 （executor）運行包含我們比特幣服務的Docker鏡像。

對這里的scheduler來說，需要運行的有兩種任務—— 單礦服務器任務和多礦服務器任務。服務器會和一個比特幣采礦池通信，并給每個“工人”分配塊。“工人”會努力工作，即開采比特幣。

任務實際上被封裝在executor框架中，因此任務運行意味著告訴Mesos master在其中一個slave上面啟動一個executor。 由于這里使用的是標準命令執行器（executor），因此可以指定任務是二進制可執行文件、bash腳本或者其他命令。由于Mesos支持 Docker，因此在本例中將使用可執行的Docker鏡像。Docker是這樣一種技術，它允許你將應用程序和它運行時需要的依賴一起打包。

為了在Mesos中使用Docker鏡像，這里需要在Docker registry中注冊它們的名稱：

const ( 
    MinerServerDockerImage = "derekchiang/p2pool" 
    MinerDaemonDockerImage = "derekchiang/cpuminer" 
) 

然后定義一個常量，指定每個任務所需資源：

const ( 
    MemPerDaemonTask = 128  // mining shouldn't be memory-intensive 
    MemPerServerTask = 256 
    CPUPerServerTask = 1    // a miner server does not use much CPU 
) 

現在定義一個真正的scheduler，對其跟蹤，并確保其正確運行需要的狀態：

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必須實現下面的接口：

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) 
} 

現在一起看一個回調函數：

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注冊之后被調用。

Reregistered在scheduler 與Mesos master斷開連接并且再次注冊時被調用，例如，在master重啟的時候。

Disconnected在scheduler 與Mesos master斷開連接時被調用。這個在master掛了的時候會發生。

目前為止，這里僅僅在回調函數中打印了日志信息，因為對于一個像這樣的簡單框架，大多數回調函數可以空在那里。然而，下一個回調函數就是每一個框架的核心，必須要認真的編寫。

ResourceOffers在scheduler 從master那里得到一個offer的時候被調用。每一個offer包含一個集群上可以給框架使用的資源列表。資源通常包括CPU、內存、端口和磁盤。一個框架可以使用它提供的一些資源、所有資源或者一點資源都不給用。

針對每一個offer，現在期望聚集所有的提供的資源并決定是否需要發布一個新的server任務或者一個新的worker任務。這里可以向每個 offer發送盡可能多的任務以測試***容量，但是由于開采比特幣是依賴CPU的，所以這里每個offer運行一個開采者任務并使用所有可用的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. 
    } 
} 

針對每個任務都需要創建一個對應的TaskInfo message ，它包含了運行這個任務需要的信息。

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指定了一些關于任務的重要元數據信息，它允許Mesos節點運行Docker容器，特別會指定name、task ID、container information以及一些需要給容器傳遞的參數。這里也會指定任務需要的資源。

現在TaskInfo已經被構建好，因此任務可以這樣運行：

driver.LaunchTasks([]*mesos.OfferID{offer.Id}, tasks, &mesos.Filters{RefuseSeconds: proto.Float64(1)})

在框架中，需要處理的***一件事情是當開采者server關閉時會發生什么。這里可以利用StatusUpdate 函數來處理。

在一個任務的生命周期中，針對不同的階段有不同類型的狀態更新。對這個框架來說，想要確保的是如果開采者server由于某種原因失敗，系統會Kill所有開采者worker以避免浪費資源。這里是相關的代碼：

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編寫分布式系統是多么快速和簡單。