實戰 Go：如何實現一個簡單分散式系統
引子
如今很多雲原生系統、分散式系統，例如 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[1]
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
如果工作節點啟動成功，將會看到如下日誌資訊。
worker node started
主節點、工作節點都啟動成功後，我們在另外一個命令列視窗中輸入如下命令來發起 API 請求。
curl -X POST \
-H "Content-Type: application/json" \
-d '{"cmd": "touch /tmp/hello-distributed-system"}' \
http://localhost:9092/tasks
在工作節點視窗應該可以看到日誌 received command: touch /tmp/hello-distributed-system。
然後檢視檔案是否順利生成，執行 ls -l /tmp/hello-distributed-system。
-rw-r--r-- 1 marvzhang wheel 0B Oct 26 12:22 /tmp/hello-distributed-system
檔案成功生成，表示已經通過工作節點執行成功了！大功告成！
總結
本篇文章通過 RPC 框架 gRPC 以及 Go 語言自帶的 Channel，將節點串接起來，開發出了一個簡單的分散式系統。所用到的核心庫和技術：
- gRPC
- Protocol Buffers
- channel
- gin
- os/exec
整個程式碼示例倉庫在 GitHub 上: https://github.com/tikazyq/codao-code/tree/main/2022-10/go-distributed-system
社群
如果您對筆者的文章感興趣，可以加筆者微信 tikazyq1 並註明 "碼之道"，筆者會將你拉入 "碼之道" 交流群。