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$
