golang實現簡易的分散式系統方法
本文介紹了golang實現簡易的分散式系統方法,分享給大家,具體如下:
功能
能夠傳送/接收請求和響應
能夠連線到叢集
如果無法連線到群集(如果它是第一個節點),則可以作為主節點啟動節點
每個節點有唯一的標識
能夠在節點之間交換json資料包
接受命令列引數中的所有資訊(將來在我們系統升級時將會很有用)
原始碼
package main import ( "fmt" "strconv" "time" "math/rand" "net" "flag" "strings" "encoding/json" ) // 節點資料資訊 type NodeInfo struct { // 節點ID,通過隨機數生成 NodeId int `json:"nodeId"` // 節點IP地址 NodeIpAddr string `json:"nodeIpAddr"` // 節點埠 Port string `json: "port"` } // 將節點資料資訊格式化輸出 //NodeInfo:{nodeId: 89423,nodeIpAddr: 127.0.0.1/8,port: 8001} func (node *NodeInfo) String() string { return "NodeInfo:{ nodeId:" + strconv.Itoa(node.NodeId) + ",nodeIpAddr:" + node.NodeIpAddr + ",port:" + node.Port + "}" } /* 新增一個節點到叢集的一個請求或者響應的標準格式 */ type AddToClusterMessage struct { // 源節點 Source NodeInfo `json:"source"` // 目的節點 Dest NodeInfo `json:"dest"` // 兩個節點連線時傳送的訊息 Message string `json:"message"` } /* Request/Response 資訊格式化輸出 */ func (req AddToClusterMessage) String() string { return "AddToClusterMessage:{\n source:" + req.Source.String() + ",\n dest: " + req.Dest.String() + ",\n message:" + req.Message + " }" } // cat vi go // rm func main() { // 解析命令列引數 makeMasterOnError := flag.Bool("makeMasterOnError", false, "如果IP地址沒有連線到叢集中,我們將其作為Master節點.") clusterip := flag.String("clusterip", "127.0.0.1:8001", "任何的節點連線都連線這個IP") myport := flag.String("myport", "8001", "ip address to run this node on. default is 8001.") flag.Parse() //解析 fmt.Println(*makeMasterOnError) fmt.Println(*clusterip) fmt.Println(*myport) /* 為節點生成ID */ rand.Seed(time.Now().UTC().UnixNano()) //種子 myid := rand.Intn(99999999) // 隨機 //fmt.Println(myid) // 獲取IP地址 myIp,_ := net.InterfaceAddrs() fmt.Println(myIp[0]) // 建立NodeInfo結構體物件 me := NodeInfo{NodeId: myid, NodeIpAddr: myIp[0].String(), Port: *myport} // 輸出結構體資料資訊 fmt.Println(me.String()) dest := NodeInfo{ NodeId: -1, NodeIpAddr: strings.Split(*clusterip, ":")[0], Port: strings.Split(*clusterip, ":")[1]} /* 嘗試連線到叢集,在已連線的情況下並且向叢集傳送請求 */ ableToConnect := connectToCluster(me, dest) /* * 監聽其他節點將要加入到叢集的請求 */ if ableToConnect || (!ableToConnect && *makeMasterOnError) { if *makeMasterOnError {fmt.Println("Will start this node as master.")} listenOnPort(me) } else { fmt.Println("Quitting system. Set makeMasterOnError flag to make the node master.", myid) } } /* * 這是傳送請求時格式化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 } 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 := "Sure buddy.. too easy.." responseMessage := getAddToClusterMessage(me, requestMessage.Source, text) json.NewEncoder(connIn).Encode(&responseMessage) connIn.Close() } } }
執行程式
/Users/liyuechun/go
liyuechun:go yuechunli$ go install main
liyuechun:go yuechunli$ main
My details: NodeInfo:{ nodeId:53163002, nodeIpAddr:127.0.0.1/8, port:8001 }
不能連線到叢集. 53163002
Quitting system. Set makeMasterOnError flag to make the node master. 53163002
liyuechun:go yuechunli$
獲取相關幫助資訊
$ ./bin/main -h
liyuechun:go yuechunli$ ./bin/main -h
Usage of ./bin/main:
-clusterip string
ip address of any node to connnect (default “127.0.0.1:8001”)
-makeMasterOnError
make this node master if unable to connect to the cluster ip provided.
-myport string
ip address to run this node on. default is 8001. (default “8001”)
liyuechun:go yuechunli$
啟動Node1主節點
$ ./bin/main --makeMasterOnError
liyuechun:go yuechunli$ ./bin/main --makeMasterOnError
My details: NodeInfo:{ nodeId:82381143, nodeIpAddr:127.0.0.1/8, port:8001 }
未連線到叢集. 82381143
Will start this node as master.
新增節點Node2到叢集
$ ./bin/main --myport 8002 --clusterip 127.0.0.1:8001
新增節點Node3到叢集
main --myport 8004 --clusterip 127.0.0.1:8001
新增節點Node4到叢集
$ main --myport 8003 --clusterip 127.0.0.1:8002