增加查找集群的代码：实现方式是集群的节点定时广播自身信息到某个广播地址端口，新节点通过该广播地址端口和密码获取该信息并解析

chenshijun

2019-08-06 514ce3d588608c96744fc72f6d8c1d3d9bbdb41f

 agent.go

@@ -18,12 +18,13 @@

import (
   "context"
   "encoding/json"
   "errors"
   "fmt"
   "github.com/hashicorp/memberlist"
   "io/ioutil"
   "net"
   "os"
   "strconv"

   //"os"
   "strings"
@@ -35,12 +36,23 @@
   "log"
)

const (
   QueryEventGetDB = "GetDatabase"
   QueryEventUpdateDBData = "UpdateDBData"
)
// Agent warps the serf agent
type Agent struct {
   *agent.Agent
   conf    *Config
   readyCh chan struct{}
   errorCh chan error
}

type NodeInfo struct {
   ClusterID string `json:"clusterID"`
   NodeID string `json:"nodeID"`
   NodeAddress  string `json:"nodeAddress"`
   IsAlive int `json:"isAlive"`
}

// Create create serf agent with config
@@ -92,6 +104,8 @@
         a.errorCh <- err
      }
   }

   go a.BroadcastMemberlist(BroadcastInterval * time.Second)
}

// HandleEvent Handles serf.EventMemberJoin events,
@@ -106,41 +120,71 @@
      var tmpstringslice []string
      tmpstringslice = append(tmpstringslice, string(ev.Payload))
      fmt.Println(tmpstringslice)
      results, err := DoExecute(tmpstringslice)
      results, err := ExecuteWriteSql(tmpstringslice)

      for _, result := range results {
         fmt.Println(result, "results err: ", err)
      }

   case *serf.Query:
      //bak file and send resp
      filename, err := BakDbFile()
      if err != nil {
         fmt.Println("bak db file error!")
         return
      }
      fmt.Println(filename)

      filebuf, err := ioutil.ReadFile(filename)
      fmt.Println("filebuf: ", len(filebuf))
      if err != nil {
         fmt.Printf("file to []bytes error: %s\n", err)
         return
      }

      err = os.Remove(filename)
      if err != nil {
         fmt.Printf("remove file%s\n failed", filename)
         return
      }

      fmt.Println("query payload: ", len(ev.Payload))
      if query, ok := event.(*serf.Query); ok {
         if err := query.Respond(filebuf); err != nil {
            fmt.Printf("err: %s\n", err)
      if ev.Name == QueryEventGetDB{
         //bak file and send resp
         filename, err := BakDbFile()
         if err != nil {
            fmt.Println("bak db file error!")
            return
         }
         fmt.Println(filename)

         filebuf, err := ioutil.ReadFile(filename)
         fmt.Println("filebuf: ", len(filebuf))
         if err != nil {
            fmt.Printf("file to []bytes error: %s\n", err)
            return
         }

         err = os.Remove(filename)
         if err != nil {
            fmt.Printf("remove file%s\n failed", filename)
            return
         }

         fmt.Println("query payload: ", len(ev.Payload))
         if query, ok := event.(*serf.Query); ok {
            if err := query.Respond(filebuf); err != nil {
               fmt.Printf("err: %s\n", err)
               return
            }
         }
      } else if ev.Name == QueryEventUpdateDBData {
         fmt.Println(string(ev.Payload))
         var tmpstringslice []string
         tmpstringslice = append(tmpstringslice, string(ev.Payload))
         fmt.Println(tmpstringslice)
         rows, err := ExecuteQuerySql(tmpstringslice)
         if err != nil {
            fmt.Println("err: ", err)
            return
         }
         var rowsReturn []Rows
         for _,r := range rows {
            rowsReturn = append(rowsReturn, *r)
         }

         bytesReturn, err := json.Marshal(rowsReturn)
         fmt.Println("results: ", bytesReturn)
         if query, ok := event.(*serf.Query); ok {
            if err := query.Respond(bytesReturn); err != nil {
               fmt.Printf("err: %s\n", err)
               return
            }
         }

         //var res []*Rows
         //json.Unmarshal(bytesReturn, &res)
      }


   default:
      fmt.Printf("Unknown event type: %s\n", ev.EventType().String())
@@ -159,6 +203,38 @@
   //}
   //a.DeregisterEventHandler(a)
   //close(a.readyCh)
}

func (a *Agent) BroadcastMemberlist(delay time.Duration) {
   //serf := a.serf
   serf := a.Agent.Serf()
   mb := serf.LocalMember()
   mblist := serf.Memberlist()
   fmt.Println("mb:", mb)

   // copy local node
   localNode := *mblist.LocalNode()
   nodeID := a.conf.NodeName
   nodeAddress := localNode.Address()
   clusterID := mb.Tags[tagKeyClusterID]
   isAlive := int(mb.Status)

   message, _ := json.Marshal(NodeInfo{
      clusterID,
      nodeID,
      nodeAddress,
      isAlive,
   })

   // replace node address
   localNode.Addr = net.ParseIP(BroadcastIP)
   //localNode.Addr = net.IPv4(255,255,255,255)
   localNode.Port = BroadcastPort
   for {
      // fmt.Printf("localNode: %v %v\n", nodeName, nodeAddress)
      mblist.SendBestEffort(&localNode, []byte(message))
      time.Sleep(delay)
   }
}

// Ready Returns a channel that will be closed when serf is ready
@@ -281,8 +357,6 @@
   return
}



//GetDbFromCluster get the newest database after join cluster
//dbPathWrite the path where to write after got a database,
func (a *Agent)GetDbFromCluster(dbPathWrite string) {
@@ -302,7 +376,7 @@
      FilterNodes: strings.Fields(specmembername),
   }

   resp, err := a.Query("getDatabase", []byte(""), &params)
   resp, err := a.Query(QueryEventGetDB, []byte(""), &params)
   if err == nil || !strings.Contains(err.Error(), "cannot contain") {
      fmt.Println("err: ", err)
   }
@@ -315,13 +389,13 @@
            fmt.Println("x length is: ", len(r.Payload))

            // // byte to file.
            Dbconn.Close()
            Dbconn = nil
            SerfDbConn.Close()
            SerfDbConn = nil
            err = ioutil.WriteFile(dbPathWrite, r.Payload, 0644)
            if err != nil {
               fmt.Println("query byte to file error!", err)
            }
            err := GetConn()
            err := InitDbConn("")
            if err != nil {
               fmt.Println("create db conn of test.db error: ", err)
            }
@@ -341,11 +415,27 @@
}

//Init serf Init
//web后台收到创建集群的请求，
func Init(clusterID string, password string, nodeID string) (*Agent, error) {
func Init(clusterID string, password string, nodeID string, ips []string) (*Agent, error) {
   agent, err := InitNode(clusterID, password, nodeID)
   if err != nil {
      fmt.Printf("InitNode failed, error: %s", err)
      return agent, err
   }

   err = agent.JoinByNodeIP(ips)
   if err != nil {
      fmt.Printf("JoinByNodeIP failed, error: %s", err)
      return agent, err
   }

   return agent, err
}

//InitNode web后台收到创建集群的请求，
func InitNode(clusterID string, password string, nodeID string) (*Agent, error) {
   conf := DefaultConfig()
   fmt.Println("clusterID:", clusterID, "password:", password, "nodeID:", nodeID)
   //conf.ClusterID = clusterID
   conf.ClusterID = clusterID
   conf.NodeName = nodeID
   if password == "" {
      conf.EncryptKey = DefaultEncryptKey
@@ -377,8 +467,18 @@
   return agent, nil
}

func (a *Agent) JoinByNodeIP(ip string) error {
   n, err := a.Agent.Join([]string{ip + ":" + strconv.Itoa(DefaultBindPort)}, true)
func (a *Agent) JoinByNodeIP(ips []string) error {
   var nodes []string

   if len(ips) == 0 {
      return fmt.Errorf("No Nodes To Join!")
   }
   for _, ip := range ips {
      node := fmt.Sprintf("%s:%d", ip, DefaultBindPort)
      nodes = append(nodes, node)
   }

   n, err := a.Agent.Join(nodes, true)
   if err != nil || n == 0{
      a.Stop()
      fmt.Println("Stop node")