Go語言實戰之實現一個簡單分散式系統

2022-10-28 14:02:04

引子

如今很多雲原生系統、分散式系統，例如 Kubernetes，都是用 Go 語言寫的，這是因為 Go 語言天然支援非同步程式設計，而且靜態語言能保證應用系統的穩定性。筆者的開源專案 Crawlab 作為爬蟲管理平臺，也應用到了分散式系統。本篇文章將介紹如何用 Go 語言編寫一個簡單的分散式系統。

思路

在開始寫程式碼之前，我們先思考一下需要實現些什麼。

主節點（Master Node）：中控系統，相當於軍隊中的指揮官，派發任務命令
工作節點（Worker Node）：執行者，相當於軍隊中的士兵，執行任務

除了上面的概念以外，我們需要實現一些簡單功能。

上報執行狀態（Report Status）：工作節點向主節點上報當前狀態
分派任務（Assign Task）：通過 API 向主節點發起請求，主節點再向工作節點分派任務
執行指令碼（Execute Script）：工作節點執行任務中的指令碼

整個流程示意圖如下。

實戰

節點通訊

節點之間的通訊在分散式系統中非常重要，畢竟每個節點或機器如果孤立執行，就失去了分散式系統的意義。因此，節點通訊在分散式系統中是核心模組。

gRPC 協定

首先，我們來想一下，如何讓節點之間進行相互通訊。最常用的通訊方式就是 API，不過這個通訊方式有個缺點，就是需要將各個節點的 IP 地址及埠顯示暴露給其他節點，這在公網中是不太安全的。因此，我們選擇了 gRPC，一種流行的遠端過程呼叫（Remote Procedure Call，RPC）框架。這裡我們不過多的解釋 RPC 或 gRPC 的原理，簡而言之，就是能讓呼叫者在遠端機器上執行命令的協定方式。

為了使用 gRPC 框架，我們先建立 go.mod 並輸入以下內容，並執行 go mod download。注意：對於國內的朋友，或許需要新增代理才能正常下載，可以先執行 export GOPROXY=goproxy.cn,direct 後再執行下載命令。

module go-distributed-system

go 1.17

require (
  github.com/golang/protobuf v1.5.0
  google.golang.org/grpc v1.27.0
  google.golang.org/protobuf v1.27.1
)

然後，我們建立 Protocol Buffers 檔案 node.proto（表示節點對應的 gRPC 協定檔案），並輸入以下內容。

syntax = "proto3";

package core;
option go_package = ".;core";

message Request {
  string action = 1;
}

message Response {
  string data = 1;
}

service NodeService {
  rpc ReportStatus(Request) returns (Response){};       // Simple RPC
  rpc AssignTask(Request) returns (stream Response){};  // Server-Side RPC
}

在這裡我們建立了兩個 RPC 服務，分別是負責上報狀態的 Simple RPC ReportStatus 以及 Server-Side RPC AssignTask。Simple RPC 和 Server-Side RPC 的區別如下圖所示，主要區別在於 Server-Side RPC 可以從通過流（Stream）向用戶端（Client）主動傳送資料，而 Simple RPC 只能從使用者端向伺服器端（Server）發請求。

建立好 .proto 檔案後，我們需要將這個 gRPC 協定檔案轉化為 .go 程式碼檔案，從而能被 Go 程式參照。在命令列視窗中執行如下命令。注意：編譯工具 protoc 不是自帶的，需要單獨下載，具體可以參考檔案 https://grpc.io/docs/protoc-installation/。

mkdir core
protoc --go_out=./core 
    --go-grpc_out=./core 
    node.proto

執行完後，可以在 core 目錄下看到兩個 Go 程式碼檔案， node.pb.go 和 node_grpc.pb.go，這相當於 Go 程式中對應的 gRPC 庫。

gRPC 伺服器端

現在開始編寫伺服器端邏輯。

咱們先建立一個新檔案 core/node_service_server.go，輸入以下內容。主要邏輯就是實現了之前建立好的 gRPC 協定中的兩個呼叫方法。其中，暴露了 CmdChannel 這個通道（Channel）來獲取需要傳送到工作節點的命令。

package core

import (
  "context"
)

type NodeServiceGrpcServer struct {
  UnimplementedNodeServiceServer

  // channel to receive command
  CmdChannel chan string
}

func (n NodeServiceGrpcServer) ReportStatus(ctx context.Context, request *Request) (*Response, error) {
  return &Response{Data: "ok"}, nil
}

func (n NodeServiceGrpcServer) AssignTask(request *Request, server NodeService_AssignTaskServer) error {
  for {
    select {
    case cmd := <-n.CmdChannel:
      // receive command and send to worker node (client)
      if err := server.Send(&Response{Data: cmd}); err != nil {
        return err
      }
    }
  }
}

var server *NodeServiceGrpcServer

// GetNodeServiceGrpcServer singleton service
func GetNodeServiceGrpcServer() *NodeServiceGrpcServer {
  if server == nil {
    server = &NodeServiceGrpcServer{
      CmdChannel: make(chan string),
    }
  }
  return server
}

gRPC 使用者端

gRPC 使用者端不需要具體實現，我們通常只需要呼叫 gRPC 使用者端的方法，程式會自動發起向伺服器端的請求以及獲取後續的響應。

主節點

編寫好了節點通訊的基礎部分，現在我們需要實現主節點了，這是整個中心化分散式系統的核心。

咱們建立一個新的檔案 node.go，輸入以下內容。

package core

import (
  "github.com/gin-gonic/gin"
  "google.golang.org/grpc"
  "net"
  "net/http"
)

// MasterNode is the node instance
type MasterNode struct {
  api     *gin.Engine            // api server
  ln      net.Listener           // listener
  svr     *grpc.Server           // grpc server
  nodeSvr *NodeServiceGrpcServer // node service
}

func (n *MasterNode) Init() (err error) {
  // TODO: implement me
  panic("implement me")
}

func (n *MasterNode) Start() {
  // TODO: implement me
  panic("implement me")
}

var node *MasterNode

// GetMasterNode returns the node instance
func GetMasterNode() *MasterNode {
  if node == nil {
    // node
    node = &MasterNode{}

    // initialize node
    if err := node.Init(); err != nil {
      panic(err)
    }
  }

  return node
}

其中，我們建立了兩個佔位方法 Init 和 Start，我們分別實現。

在初始化方法 Init 中，我們需要做幾件事情：

註冊 gRPC 服務
註冊 API 服務

現在，在 Init 方法中加入如下程式碼。

func (n *MasterNode) Init() (err error) {
  // grpc server listener with port as 50051
  n.ln, err = net.Listen("tcp", ":50051")
  if err != nil {
    return err
  }

  // grpc server
  n.svr = grpc.NewServer()

  // node service
  n.nodeSvr = GetNodeServiceGrpcServer()

  // register node service to grpc server
  RegisterNodeServiceServer(node.svr, n.nodeSvr)

  // api
  n.api = gin.Default()
  n.api.POST("/tasks", func(c *gin.Context) {
    // parse payload
    var payload struct {
      Cmd string `json:"cmd"`
    }
    if err := c.ShouldBindJSON(&payload); err != nil {
      c.AbortWithStatus(http.StatusBadRequest)
      return
    }

    // send command to node service
    n.nodeSvr.CmdChannel <- payload.Cmd

    c.AbortWithStatus(http.StatusOK)
  })

  return nil
}

可以看到，我們新建了一個 gRPC Server，並將之前的 NodeServiceGrpcServer 註冊了進去。另外，我們還用 gin 框架建立了一個簡單的 API 服務，可以 POST 請求到 /tasks 向 NodeServiceGrpcServer 中的命令通道 CmdChannel 傳送命令。這樣就將各個部件串接起來了！

啟動方法 Start 很簡單，就是啟動 gRPC Server 以及 API Server。

func (n *MasterNode) Start() {
  // start grpc server
  go n.svr.Serve(n.ln)

  // start api server
  _ = n.api.Run(":9092")

  // wait for exit
  n.svr.Stop()
}

下一步，我們就要實現實際做任務的工作節點了。

工作節點

現在，我們建立一個新檔案 core/worker_node.go，輸入以下內容。

package core

import (
  "context"
  "google.golang.org/grpc"
  "os/exec"
)

type WorkerNode struct {
  conn *grpc.ClientConn  // grpc client connection
  c    NodeServiceClient // grpc client
}

func (n *WorkerNode) Init() (err error) {
  // connect to master node
  n.conn, err = grpc.Dial("localhost:50051", grpc.WithInsecure())
  if err != nil {
    return err
  }

  // grpc client
  n.c = NewNodeServiceClient(n.conn)

  return nil
}

func (n *WorkerNode) Start() {
  // log
  fmt.Println("worker node started")

  // report status
  _, _ = n.c.ReportStatus(context.Background(), &Request{})

  // assign task
  stream, _ := n.c.AssignTask(context.Background(), &Request{})
  for {
    // receive command from master node
    res, err := stream.Recv()
    if err != nil {
      return
    }

    // log command
    fmt.Println("received command: ", res.Data)

    // execute command
    parts := strings.Split(res.Data, " ")
    if err := exec.Command(parts[0], parts[1:]...).Run(); err != nil {
      fmt.Println(err)
    }
  }
}

var workerNode *WorkerNode

func GetWorkerNode() *WorkerNode {
  if workerNode == nil {
    // node
    workerNode = &WorkerNode{}

    // initialize node
    if err := workerNode.Init(); err != nil {
      panic(err)
    }
  }

  return workerNode
}

其中，我們在初始化方法 Init 中建立了gRPC 使用者端，並連線了主節點的 gRPC 伺服器端。

在啟動方法 Start 中做了幾件事情：

呼叫上報狀態（Report Status）的 Simple RPC 方法
呼叫分配任務（Assign Task）的 Server-Side RPC 方法，獲取到了流（Stream）
通過迴圈不斷接受流傳輸過來的來自伺服器端（也就是主節點）的資訊，並執行命令

這樣，整個包含主節點、工作節點的分散式系統核心邏輯就寫好了！

將它們放在一起

最後，我們需要將這些核心邏輯用命令列工具封裝一下，以便啟用。

建立主程式檔案 main.go，並輸入以下內容。

package main

import (
  "go-distributed-system/core"
  "os"
)

func main() {
  nodeType := os.Args[0]
  switch nodeType {
  case "master":
    core.GetMasterNode().Start()
  case "worker":
    core.GetWorkerNode().Start()
  default:
    panic("invalid node type")
  }
}

這樣，整個簡單的分散式系統就建立好了！

程式碼效果

下面我們來執行一下程式碼。

開啟兩個命令列視窗，其中一個輸入 go run main.go master 啟動主節點，另一個輸入 go run main.go worker 啟動工作節點。

如果主節點啟動成功，將會看到如下紀錄檔資訊。

[GIN-debug] [WARNING] Creating an Engine instance with the Logger and Recovery middleware already attached.

[GIN-debug] [WARNING] Running in "debug" mode. Switch to "release" mode in production.
 - using env:   export GIN_MODE=release
 - using code:  gin.SetMode(gin.ReleaseMode)

[GIN-debug] POST   /tasks                    --> go-distributed-system/core.(*MasterNode).Init.func1 (3 handlers)
[GIN-debug] [WARNING] You trusted all proxies, this is NOT safe. We recommend you to set a value.
Please check https://pkg.go.dev/github.com/gin-gonic/gin#readme-don-t-trust-all-proxies for details.
[GIN-debug] Listening and serving HTTP on :9092

如果工作節點啟動成功，將會看到如下紀錄檔資訊。