【转】golang实现简易的分布式系统方法

转， 原文：https://www.jb51.net/article/148425.htm

--------

本文介绍了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:{
   source:" + req.Source.String() + ",
   dest: " + req.Dest.String() + ",
   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("得到数据响应:
  " + 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:
  " + requestMessage.String())
  
        text := "Sure buddy.. too easy.."
        responseMessage := getAddToClusterMessage(me, requestMessage.Source, text)
        json.NewEncoder(connIn).Encode(&responseMessage)
        connIn.Close()
      }
    }
  }
  

  　　

  运行程序

  1 2 3 4 5 6 7

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

  获取相关帮助信息

  1	$ ./bin/main -h

  1 2 3 4 5 6 7 8 9

  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主节点

  1	$ ./bin/main --makeMasterOnError

  1 2 3 4

  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到集群

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

  添加节点Node3到集群

  1	main --myport 8004 --clusterip 127.0.0.1:8001

  添加节点Node4到集群

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