当前位置: 移动技术网 > IT编程>脚本编程>Go语言 > golang实现简易的分布式系统方法

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

2018年11月08日  | 移动技术网IT编程  | 我要评论

本文介绍了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

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持移动技术网。

如对本文有疑问, 点击进行留言回复!!

相关文章:

验证码:
移动技术网