Go语言实战之实现一个简单分布式系统
作者:MarvinZhang 发布时间:2024-05-05 09:33:56
引子
如今很多云原生系统、分布式系统,例如 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
如果工作节点启动成功,将会看到如下日志信息。
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
文件成功生成,表示已经通过工作节点执行成功了!大功告成!
来源:https://juejin.cn/post/7158685917006266398


猜你喜欢
- 阅读上一篇:javascript面向对象编程(三)继承是面向对象语言中的一个重要概念,现在我们来探讨一下继承。在网上搜一下javascrip
- 一:mysql官网下载https://dev.mysql.com/downloads/file/?id=494993
- pygame.transform 模块允许您对加载、创建后的图像进行一系列操作,比如调整图像大小、旋转图片等操作,常用方法如下所示:下面看一
- 上节回顾主要讲了协程、进程、异步IO多路复用。 协程和IO多路复用都是单线程的。epoll 在linux下通过这个模块libev
- 线性回归在整个财务中广泛应用于众多应用程序中。在之前的教程中,我们使用普通最小二乘法(OLS)计算了公司的beta与相对索引的比较。现在,我
- Dubbo提供动态调整超时时间的服务治理能力,可以在无需重启应用的情况下,动态调整服务超时时间。Dubbo可以通过XML配置,注解配置,动态
- python默认使用的是国外镜像,有时候下载非常慢,最快的办法就是在下载命令中增加国内源:常用的国内源如下:清华大学:https://pyp
- PyTorch中还单独提供了一个sampler模块,用来对数据进行采样。常用的有随机采样器:RandomSampler,当dataloade
- 我之前想写路由器的密码暴力破解器,我手上只有极路由,发现极路由有安全限制,只能允许连续10密码错误,所以我改拿博客园练手。博客园的博客有个功
- 长期以来我就有对几年来交互设计的心得进行总结整理的想法。回到中国来亲身体会到不少同行,主要是交互设计师和视觉设计师对于交互设计的困惑,以及其
- 1、例子:拟合一种函数Func,此处为一个指数函数。出处:SciPy v1.1.0 Reference Guide#Headerimport
- 这个模块提供了与 Perl 相似l的正则表达式匹配操作。Unicode字符串也同样适用。正则表达式使用反斜杠" \ "来
- 1、前期准备通过 pip 或 easy_install 安装了 pymongo 之后, 就能通过 Python 调教 mongodb 了.接
- 元数据简介元数据 (metadata) 最常见的定义为“有关数据的结构数据”,或者再简单一点就是“关于数据的信息”,日常生活中的图例、图书馆
- 首先澄清一个应用场景问题。研究(1)中指出,对于结构复杂的网站,不少设计师们喜欢采用960固定宽度布局。但要注意的是,960并不是万能钥匙,
- 在 HTML 页面中嵌入 JavaScript 脚本需要使用 <script> 标签,用户可以在 <script>
- 如果你的PHP网站换了空间,必定要对Mysql数据库进行转移,一般的转移的方法,是备份再还原,有点繁琐,而且由于数据库版本的不一样会导致数据
- 简介前几天捣鼓了一下Ubuntu,正是想用一下我旧电脑上的N卡,可以用GPU来跑代码,体验一下多核的快乐。还好我这破电脑也是支持Cuda的:
- 1. 前言中文分词≠自然语言处理!HanlpHanLP是由一系列模型与算法组成的Java工具包,目标是普及自然语言处理在生产环境中的应用。H
- Token验证是验证用户身份的重要方式,在golang开发中具有广泛应用,文中主要阐述了利用jwt包加密后的token验证。导入包:impo