如何通过memberlist库实现gossip管理集群及集群数据交互问题

package mainimport ("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 clusterfor _, 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                DelegateEvents                  EventDelegateConflict                ConflictDelegateMerge                   MergeDelegatePing                    PingDelegateAlive                   AliveDelegate//...}

const (pingMsg messageType = iotaindirectPingMsgackRespMsgsuspectMsgaliveMsgdeadMsgpushPullMsgcompoundMsguserMsg // User mesg, not handled by uscompressMsgencryptMsgnackRespMsghasCrcMsgerrMsg)

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 neededNotifyMsg([]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)}

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 detectedNotifyConflict(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 ackAckPayload() []byte// NotifyPing is invoked when an ack for a ping is receivedNotifyPingComplete(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 broadcastInvalidates(b Broadcast) bool// Returns a byte form of the messageMessage() []byte// Finished is invoked when the message will no longer// be broadcast, either due to invalidation or to the// transmit limit being reachedFinished()}

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

type simpleBroadcast []bytefunc (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 intmu    sync.Mutextq    *btree.BTree // stores *limitedBroadcast as btree.Itemtm    map[string]*limitedBroadcastidGen 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.DurationGossipNodes         intGossipToTheDeadTime 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}