golang實(shí)現(xiàn)簡易的分布式系統(tǒng)方法

package main

import (
  "fmt"
  "strconv"
  "time"
  "math/rand"
  "net"
  "flag"
  "strings"
  "encoding/json"
)

// 節(jié)點(diǎn)數(shù)據(jù)信息
type NodeInfo struct {

  // 節(jié)點(diǎn)ID，通過隨機(jī)數(shù)生成
  NodeId int `json:"nodeId"`
  // 節(jié)點(diǎn)IP地址
  NodeIpAddr string `json:"nodeIpAddr"`
  // 節(jié)點(diǎn)端口
  Port string `json: "port"`
}

// 將節(jié)點(diǎn)數(shù)據(jù)信息格式化輸出
//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 + "}"
}

/* 添加一個(gè)節(jié)點(diǎn)到集群的一個(gè)請求或者響應(yīng)的標(biāo)準(zhǔn)格式 */
type AddToClusterMessage struct {
  // 源節(jié)點(diǎn)
  Source NodeInfo `json:"source"`
  // 目的節(jié)點(diǎn)
  Dest NodeInfo `json:"dest"`
  // 兩個(gè)節(jié)點(diǎn)連接時(shí)發(fā)送的消息
  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() {

  // 解析命令行參數(shù)
  makeMasterOnError := flag.Bool("makeMasterOnError", false, "如果IP地址沒有連接到集群中，我們將其作為Master節(jié)點(diǎn).")
  clusterip := flag.String("clusterip", "127.0.0.1:8001", "任何的節(jié)點(diǎn)連接都連接這個(gè)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)

  /* 為節(jié)點(diǎn)生成ID */
  rand.Seed(time.Now().UTC().UnixNano()) //種子
  myid := rand.Intn(99999999) // 隨機(jī)

  //fmt.Println(myid)

  // 獲取IP地址
  myIp,_ := net.InterfaceAddrs()
  fmt.Println(myIp[0])

  // 創(chuàng)建NodeInfo結(jié)構(gòu)體對象
  me := NodeInfo{NodeId: myid, NodeIpAddr: myIp[0].String(), Port: *myport}
  // 輸出結(jié)構(gòu)體數(shù)據(jù)信息
  fmt.Println(me.String())
  dest := NodeInfo{ NodeId: -1, NodeIpAddr: strings.Split(*clusterip, ":")[0], Port: strings.Split(*clusterip, ":")[1]}

  /* 嘗試連接到集群，在已連接的情況下并且向集群發(fā)送請求 */
  ableToConnect := connectToCluster(me, dest)

  /*
   * 監(jiān)聽其他節(jié)點(diǎn)將要加入到集群的請求
   */
  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)
  }

}

/*
 * 這是發(fā)送請求時(shí)格式化json包有用的工具
 * 這是非常重要的，如果不經(jīng)過數(shù)據(jù)格式化，你最終發(fā)送的將是空白消息
 */
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的相關(guān)細(xì)節(jié)信息 */
  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("連接到集群. 發(fā)送消息到節(jié)點(diǎn).")
    text := "Hi nody.. 請?zhí)砑游业郊?."
    requestMessage := getAddToClusterMessage(me, dest, text)
    json.NewEncoder(connOut).Encode(requestMessage)

    decoder := json.NewDecoder(connOut)
    var responseMessage AddToClusterMessage
    decoder.Decode(responseMessage)
    fmt.Println("得到數(shù)據(jù)響應(yīng):\n" + responseMessage.String())

    return true
  }
  return false
}

func listenOnPort(me NodeInfo){
  /* 監(jiān)聽即將到來的消息 */
  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$

$ ./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.

$ ./bin/main --myport 8002 --clusterip 127.0.0.1:8001

main --myport 8004 --clusterip 127.0.0.1:8001

$ main --myport 8003 --clusterip 127.0.0.1:8002