通过memberlist库实现gossip管理集群以及集群数据交互

package main
import (
  "fmt"
  "github.com/hashicorp/memberlist"
  "time"
)
func main() {
  /* Create the initial memberlist from a safe configuration.
     Please reference the godoc for other default config types.
     http://godoc.org/github.com/hashicorp/memberlist#Config
  */
  list, err := memberlist.Create(memberlist.DefaultLocalConfig())
  if err != nil {
    panic("Failed to create memberlist: " + err.Error())
  }
  t := time.NewTicker(time.Second * 5)
  for {
    select {
    case <-t.C:
      // Join an existing cluster by specifying at least one known member.
      n, err := list.Join([]string{"192.168.80.129"})
      if err != nil {
        fmt.Println("Failed to join cluster: " + err.Error())
        continue
      }
      fmt.Println("member number is:", n)
      goto END
    }
  }
END:
  for {
    select {
    case <-t.C:
      // Ask for members of the cluster
      for _, member := range list.Members() {
        fmt.Printf("Member: %s %s\n", member.Name, member.Addr)
      }
    }
  }
  // Continue doing whatever you need, memberlist will maintain membership
  // information in the background. Delegates can be used for receiving
  // events when members join or leave.
}

type Config struct {
  // ...
  // Delegate and Events are delegates for receiving and providing
  // data to memberlist via callback mechanisms. For Delegate, see
  // the Delegate interface. For Events, see the EventDelegate interface.
  //
  // The DelegateProtocolMin/Max are used to guarantee protocol-compatibility
  // for any custom messages that the delegate might do (broadcasts,
  // local/remote state, etc.). If you don't set these, then the protocol
  // versions will just be zero, and version compliance won't be done.
  Delegate                Delegate
  Events                  EventDelegate
  Conflict                ConflictDelegate
  Merge                   MergeDelegate
  Ping                    PingDelegate
  Alive                   AliveDelegate
  //...
}

const (
  pingMsg messageType = iota
  indirectPingMsg
  ackRespMsg
  suspectMsg
  aliveMsg
  deadMsg
  pushPullMsg
  compoundMsg
  userMsg // User mesg, not handled by us
  compressMsg
  encryptMsg
  nackRespMsg
  hasCrcMsg
  errMsg
)

type Delegate interface {
  // NodeMeta is used to retrieve meta-data about the current node
  // when broadcasting an alive message. It's length is limited to
  // the given byte size. This metadata is available in the Node structure.
  NodeMeta(limit int) []byte
  // NotifyMsg is called when a user-data message is received.
  // Care should be taken that this method does not block, since doing
  // so would block the entire UDP packet receive loop. Additionally, the byte
  // slice may be modified after the call returns, so it should be copied if needed
  NotifyMsg([]byte)
  // GetBroadcasts is called when user data messages can be broadcast.
  // It can return a list of buffers to send. Each buffer should assume an
  // overhead as provided with a limit on the total byte size allowed.
  // The total byte size of the resulting data to send must not exceed
  // the limit. Care should be taken that this method does not block,
  // since doing so would block the entire UDP packet receive loop.
  GetBroadcasts(overhead, limit int) [][]byte
  // LocalState is used for a TCP Push/Pull. This is sent to
  // the remote side in addition to the membership information. Any
  // data can be sent here. See MergeRemoteState as well. The `join`
  // boolean indicates this is for a join instead of a push/pull.
  LocalState(join bool) []byte
  // MergeRemoteState is invoked after a TCP Push/Pull. This is the
  // state received from the remote side and is the result of the
  // remote side's LocalState call. The 'join'
  // boolean indicates this is for a join instead of a push/pull.
  MergeRemoteState(buf []byte, join bool)
}

//使用UDP方式将用户消息传输到给定节点，消息大小受限于memberlist的UDPBufferSize配置。没有使用gossip机制
func (m *Memberlist) SendBestEffort(to *Node, msg []byte) error
//与SendBestEffort机制相同，只不过一个指定了Node，一个指定了Node地址
func (m *Memberlist) SendToAddress(a Address, msg []byte) error
//使用TCP方式将用户消息传输到给定节点，消息没有大小限制。没有使用gossip机制
func (m *Memberlist) SendReliable(to *Node, msg []byte) error

type EventDelegate interface {
  // NotifyJoin is invoked when a node is detected to have joined.
  // The Node argument must not be modified.
  NotifyJoin(*Node)
  // NotifyLeave is invoked when a node is detected to have left.
  // The Node argument must not be modified.
  NotifyLeave(*Node)
  // NotifyUpdate is invoked when a node is detected to have
  // updated, usually involving the meta data. The Node argument
  // must not be modified.
  NotifyUpdate(*Node)
}

type ChannelEventDelegate struct {
Ch chan<- NodeEvent
}

type ConflictDelegate interface {
  // NotifyConflict is invoked when a name conflict is detected
  NotifyConflict(existing, other *Node)
}

type MergeDelegate interface {
  // NotifyMerge is invoked when a merge could take place.
  // Provides a list of the nodes known by the peer. If
  // the return value is non-nil, the merge is canceled.
  NotifyMerge(peers []*Node) error
}

type PingDelegate interface {
  // AckPayload is invoked when an ack is being sent; the returned bytes will be appended to the ack
  AckPayload() []byte
  // NotifyPing is invoked when an ack for a ping is received
  NotifyPingComplete(other *Node, rtt time.Duration, payload []byte)
}

type AliveDelegate interface {
  // NotifyAlive is invoked when a message about a live
  // node is received from the network.  Returning a non-nil
  // error prevents the node from being considered a peer.
  NotifyAlive(peer *Node) error
}

// Broadcast is something that can be broadcasted via gossip to
// the memberlist cluster.
type Broadcast interface {
  // Invalidates checks if enqueuing the current broadcast
  // invalidates a previous broadcast
  Invalidates(b Broadcast) bool
  // Returns a byte form of the message
  Message() []byte
  // Finished is invoked when the message will no longer
  // be broadcast, either due to invalidation or to the
  // transmit limit being reached
  Finished()
}

func (q *TransmitLimitedQueue) QueueBroadcast(b Broadcast) {
  q.queueBroadcast(b, 0)
}

type simpleBroadcast []byte
func (b simpleBroadcast) Message() []byte                       { return []byte(b) }
func (b simpleBroadcast) Invalidates(memberlist.Broadcast) bool { return false }
func (b simpleBroadcast) Finished()

// TransmitLimitedQueue is used to queue messages to broadcast to
// the cluster (via gossip) but limits the number of transmits per
// message. It also prioritizes messages with lower transmit counts
// (hence newer messages).
type TransmitLimitedQueue struct {
  // NumNodes returns the number of nodes in the cluster. This is
  // used to determine the retransmit count, which is calculated
  // based on the log of this.
  NumNodes func() int
  // RetransmitMult is the multiplier used to determine the maximum
  // number of retransmissions attempted.
  RetransmitMult int
  mu    sync.Mutex
  tq    *btree.BTree // stores *limitedBroadcast as btree.Item
  tm    map[string]*limitedBroadcast
  idGen int64
}

// GossipInterval and GossipNodes are used to configure the gossip
  // behavior of memberlist.
  //
  // GossipInterval is the interval between sending messages that need
  // to be gossiped that haven't been able to piggyback on probing messages.
  // If this is set to zero, non-piggyback gossip is disabled. By lowering
  // this value (more frequent) gossip messages are propagated across
  // the cluster more quickly at the expense of increased bandwidth.
  //
  // GossipNodes is the number of random nodes to send gossip messages to
  // per GossipInterval. Increasing this number causes the gossip messages
  // to propagate across the cluster more quickly at the expense of
  // increased bandwidth.
  //
  // GossipToTheDeadTime is the interval after which a node has died that
  // we will still try to gossip to it. This gives it a chance to refute.
  GossipInterval      time.Duration
  GossipNodes         int
  GossipToTheDeadTime time.Duration

func (c *Channel) Broadcast(b []byte) {
  b, err := proto.Marshal(&clusterpb.Part{Key: c.key, Data: b})
  if err != nil {
    return
  }
  if OversizedMessage(b) {
    select {
    case c.msgc <- b: //从c.msgc 接收数据，并使用SendReliable发送
    default:
      level.Debug(c.logger).Log("msg", "oversized gossip channel full")
      c.oversizeGossipMessageDroppedTotal.Inc()
    }
  } else {
    c.send(b)
  }
}
func OversizedMessage(b []byte) bool {
  return len(b) > MaxGossipPacketSize/2
}