go 实现分布式系统
使用go语言写一个简单的分布式系统，具有以下功能：
- 能够发送和接收请求/响应
- 能够连接到集群
- 如果无法连接到集群（假设它是第一个节点），则可以作为主节点启动节点
- 每个节点有唯一的标识
- 能够在节点之间交换json数据包
- 接收命令行参数的所有信息
一. 搭建模型
首先我们需要一个存储节点信息的结构体，包括节点id、节点ip、节点端口号：
//结构体对象可以和json互转
type NodeInfo struct {
NodeId int `json:"nodeId"` //节点ID，通过随机数生成
NodeIpAddr string `json:"nodeIpAddr"` //节点ip地址
Port string `json:"port"` //节点端口号
}
其次我们需要一个节点到集群的一个请求或者响应的标准格式的结构体
type AddToClusterMessage struct {
Source NodeInfo `json:"source"`
Dest NodeInfo `json:"dest"`
Message string `json:"message"`
}
然后分别实现结构体的tostring格式化方法，整体代码distribute.go如下：
package main
import (
"fmt"
"math/rand"
"net"
"strconv"
"time"
)
//用于json和结构体对象的互转
type NodeInfo struct {
NodeId int `json:"nodeId"` //节点ID，通过随机数生成
NodeIpAddr string `json:"nodeIpAddr"` //节点ip地址
Port string `json:"port"` //节点端口号
}
//添加一个节点到集群的一个请求或者响应的标准格式
type AddToClusterMessage struct {
Source NodeInfo `json:"source"`
Dest NodeInfo `json:"dest"`
Message string `json:"message"`
}
//将节点信息格式化输出
func (node *NodeInfo) String() string {
return "NodeInfo {nodeId:" + strconv.Itoa(node.NodeId) + ", nodeIpAddr:" + node.NodeIpAddr + ", port:" + node.Port + "}"
}
//将添加节点信息格式化
func (req AddToClusterMessage) String() string {
return "AddToClusterMessage:{\n source:" + req.Source.String() + ",\n dest: " + req.Dest.String() + ",\n message:" + req.Message + " }"
}
func main() {
// makeMasterOnError := flag.Bool("makeMasterOnError", false, "make this node master if unable to connect to the cluster ip provided.")
// clusterip := flag.String("clusterip", "127.0.0.1:8001", "ip address of any node to connnect")
// myport := flag.String("myport", "8001", "ip address to run this node on. default is 8001."
rand.Seed(time.Now().UTC().UnixNano()) //种子
myid := rand.Intn(9999999)
fmt.Println(myid)
//获取ip地址
myIp, _ := net.InterfaceAddrs()
fmt.Println(myIp[13])
//创建nodeInfo结构体
me := NodeInfo{NodeId: myid, NodeIpAddr: myIp[13].String(), Port: "8001"}
fmt.Println(me.String())
}
然后跑一下代码：
go run distribute.go
输出如图： 
二. 接受命令行参数解析
我们使用flag模块来解析命令行参数，先看一个简单的例子了解下flag：
package main
import (
"flag"
"fmt"
)
func main() {
//第一个参数，为参数名称，第二个参数为默认值，第三个参数是说明
username := flag.String("name", "", "Input your username")
flag.Parse()
fmt.Println("Hello, ", *username)
}
编译：
go build flag.go
运行：
./flag -name=world
输出：
Hello, world
如果不输入name参数：
./flag
则输出：
Hello,
然后在这个简单的分布式系统里面我们要做什么，有三个命令需要解析：
//当第一个节点启动用这个命令来将第一个节点作为主节点
makeMasterOnError := flag.Bool("makeMasterOnError", false, "make this node master if unable to connect to the cluster ip provided.")
//设置要连接的目的地ip地址
clusterip := flag.String("clusterip", "127.0.0.1:8001", "ip address of any node to connnect")
//设置要连接的目的地端口号
myport := flag.String("myport", "8001", "ip address to run this node on. default is 8001.")
flag.Parse()
三. 连接函数以及监听函数
连接目的地节点ip地址以及端口号：
func connectToCluster(me NodeInfo, dest NodeInfo) bool {
//连接到socket的相关细节信息
connOut, err := net.DialTimeout("tcp", dest.NodeIpAddr+":"+dest.Port, time.Duration(10)*time.Second)
if err != nil {
if _, ok := err.(net.Error); ok {
fmt.Println("不能连接到集群", me.NodeId)
return false
}
} else {
fmt.Println("连接到集群")
text := "Hi nody.. 请添加我到集群"
requestMessage := getAddToClusterMessage(me, dest, text)
json.NewEncoder(connOut).Encode(&requestMessage)
decoder := json.NewDecoder(connOut)
var responseMessage AddToClusterMessage
decoder.Decode(&responseMessage)
fmt.Println("得到数据响应:\n" + responseMessage.String())
return true
}
return false
}
getAddToClusterMessage函数用来返回响应信息：
//发送请求时格式化json包有用的工具
func getAddToClusterMessage(source NodeInfo, dest NodeInfo, message string) AddToClusterMessage {
return AddToClusterMessage{
Source: NodeInfo{
NodeId: source.NodeId,
NodeIpAddr: source.NodeIpAddr,
Port: source.Port},
Dest: NodeInfo{
NodeId: dest.NodeId,
NodeIpAddr: dest.NodeIpAddr,
Port: dest.Port},
Message: message,
}
}
本节点连接目的地节点成功或者成为第一个主节点后需要开启监听：
//me节点连接其它节点成功或者自身成为主节点之后开始监听别的节点在未来可能对它自身的连接
func listenOnPort(me NodeInfo) {
//监听即将到来的信息
ln, _ := net.Listen("tcp", fmt.Sprint(":"+me.Port))
//接受连接
for {
connIn, err := ln.Accept()
if err != nil {
if _, ok := err.(net.Error); ok {
fmt.Println("Error received while listening.", me.NodeId)
}
} else {
var requestMessage AddToClusterMessage
json.NewDecoder(connIn).Decode(&requestMessage)
fmt.Println("Got request:\n" + requestMessage.String())
text := "已添加你到集群"
responseMessage := getAddToClusterMessage(me, requestMessage.Source, text)
json.NewEncoder(connIn).Encode(&responseMessage)
connIn.Close()
}
}
}
四. 完整代码
package main
import (
"encoding/json"
"flag"
"fmt"
"math/rand"
"net"
"strconv"
"strings"
"time"
)
//用于json和结构体对象的互转
type NodeInfo struct {
NodeId int `json:"nodeId"` //节点ID，通过随机数生成
NodeIpAddr string `json:"nodeIpAddr"` //节点ip地址
Port string `json:"port"` //节点端口号
}
//添加一个节点到集群的一个请求或者响应的标准格式
type AddToClusterMessage struct {
Source NodeInfo `json:"source"`
Dest NodeInfo `json:"dest"`
Message string `json:"message"`
}
//将节点信息格式化输出
func (node *NodeInfo) String() string {
return "NodeInfo {nodeId:" + strconv.Itoa(node.NodeId) + ", nodeIpAddr:" + node.NodeIpAddr + ", port:" + node.Port + "}"
}
//将添加节点信息格式化
func (req AddToClusterMessage) String() string {
return "AddToClusterMessage:{\n source:" + req.Source.String() + ",\n dest: " + req.Dest.String() + ",\n message:" + req.Message + " }"
}
func main() {
makeMasterOnError := flag.Bool("makeMasterOnError", false, "make this node master if unable to connect to the cluster ip provided.")
clusterip := flag.String("clusterip", "127.0.0.1:8001", "ip address of any node to connnect")
myport := flag.String("myport", "8001", "ip address to run this node on. default is 8001.")
flag.Parse()
rand.Seed(time.Now().UTC().UnixNano()) //种子
myid := rand.Intn(9999999)
//获取ip地址
myIp, _ := net.InterfaceAddrs()
//创建nodeInfo结构体
me := NodeInfo{NodeId: myid, NodeIpAddr: myIp[13].String(), Port: *myport}
dest := NodeInfo{NodeId: -1, NodeIpAddr: strings.Split(*clusterip, ":")[0], Port: strings.Split(*clusterip, ":")[1]}
fmt.Println("我的节点信息：", me.String())
//尝试连接到集群，在已连接的情况下向集群发送请求
ableToConnect := connectToCluster(me, dest)
//如果dest节点不存在，则me节点为主节点启动，否则直接退出系统
if ableToConnect || (!ableToConnect && *makeMasterOnError) {
if *makeMasterOnError {
fmt.Println("将启动me节点为主节点")
}
listenOnPort(me)
} else {
fmt.Println("正在退出系统，请设置me节点为主节点")
}
}
//发送请求时格式化json包有用的工具
func getAddToClusterMessage(source NodeInfo, dest NodeInfo, message string) AddToClusterMessage {
return AddToClusterMessage{
Source: NodeInfo{
NodeId: source.NodeId,
NodeIpAddr: source.NodeIpAddr,
Port: source.Port},
Dest: NodeInfo{
NodeId: dest.NodeId,
NodeIpAddr: dest.NodeIpAddr,
Port: dest.Port},
Message: message,
}
}
func connectToCluster(me NodeInfo, dest NodeInfo) bool {
//连接到socket的相关细节信息
connOut, err := net.DialTimeout("tcp", dest.NodeIpAddr+":"+dest.Port, time.Duration(10)*time.Second)
if err != nil {
if _, ok := err.(net.Error); ok {
fmt.Println("不能连接到集群", me.NodeId)
return false
}
} else {
fmt.Println("连接到集群")
text := "Hi nody.. 请添加我到集群"
requestMessage := getAddToClusterMessage(me, dest, text)
json.NewEncoder(connOut).Encode(&requestMessage)
decoder := json.NewDecoder(connOut)
var responseMessage AddToClusterMessage
decoder.Decode(&responseMessage)
fmt.Println("得到数据响应:\n" + responseMessage.String())
return true
}
return false
}
//me节点连接其它节点成功或者自身成为主节点之后开始监听别的节点在未来可能对它自身的连接
func listenOnPort(me NodeInfo) {
//监听即将到来的信息
ln, _ := net.Listen("tcp", fmt.Sprint(":"+me.Port))
//接受连接
for {
connIn, err := ln.Accept()
if err != nil {
if _, ok := err.(net.Error); ok {
fmt.Println("Error received while listening.", me.NodeId)
}
} else {
var requestMessage AddToClusterMessage
json.NewDecoder(connIn).Decode(&requestMessage)
fmt.Println("Got request:\n" + requestMessage.String())
text := "已添加你到集群"
responseMessage := getAddToClusterMessage(me, requestMessage.Source, text)
json.NewEncoder(connIn).Encode(&responseMessage)
connIn.Close()
}
}
}
go install之后启动第一个节点：
main --makeMasterOnError true
另开一个终端连接刚刚启动的主节点8001：
main --myport 8002 --clusterip 127.0.0.1:8001
可以发现得到8001节点的响应，8001节点也得到8002的请求
当然也可以使用8003节点去连接8002节点：
main --myport 8003 --clusterip 127.0.0.1:8002
2021年 Stack Overflow 的调查报告显示，Rust 名列最受欢迎编程语言的榜首，86% 的开发人员表示今后会继续使用该语言。自 2016 年以来，Rust 一直在该调查报告中名列前茅。Tiobe ...
