gossip协议的原理和实战应⽤
先理解⼀下gossip协议：在⼀个有界⽹络中，每个节点都随机地与其他节点通信，经过⼀番杂乱⽆章的通信，最终所有节点的状态都会达成⼀致。每个节点可能知道所有其他节点，也可能仅知道⼏个邻居节点，只要这些节可以通过⽹络连通，最终他们的状态都是⼀致的，当然这也是疫情传播的特点。
简单的描述下这个协议，⾸先要传播谣⾔就要有种⼦节点。种⼦节点每秒都会随机向其他节点发送⾃⼰所拥有的节点列表，以及需要传播的消息。任何新加⼊的节点，就在这种传播⽅式下很快地被全⽹所知道。这个协议的神奇就在于它从设计开始就没想到信息⼀定要传递给所有的节点，但是随着时间的增长，在最终的某⼀时刻，全⽹会得到相同的信息。当然这个时刻可能仅仅存在于理论，永远不可达。
这个算法已经有⼤神实现了，参见
我这⾥先通过⼀个demo演⽰⼀下，怎么使⽤
package main
import (
"encoding/json"
"flag"
"fmt"
"net/http"
"os"
"strings"
"sync"
"/hashicorp/memberlist"
"/pborman/uuid"
)
var (
mtx sync.RWMutex
members = flag.String("members", "", "comma perated list of members")
port = flag.Int("port", 4001, "http port")
items = map[string]string{}
broadcasts *memberlist.TransmitLimitedQueue
)
type broadcast struct {
msg []byte
notify chan<- struct{}
}
type delegate struct{}
type update struct {
Action string // add, del
Data map[string]string
}
func init() {
flag.Par()
}
func (b *broadcast) Invalidates(other memberlist.Broadcast) bool {
return fal
}
func (b *broadcast) Message() []byte {
return b.msg
}
func (b *broadcast) Finished() {
ify != nil {
文明出行ify)
}
}
func (d *delegate) NodeMeta(limit int) []byte {
return []byte{}
}
func (d *delegate) NotifyMsg(b []byte) {
个人所得税怎么算if len(b) == 0 {
return
}
switch b[0] {
ca'd': // data
var updates []*update
if err := json.Unmarshal(b[1:], &updates); err != nil {
return
}
mtx.Lock()
for _, u := range updates {
for k, v := range u.Data {
switch u.Action {
ca"add":
items[k] = v
ca"del":
delete(items, k)
}
}
}
mtx.Unlock()
}
}
func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte { return broadcasts.GetBroadcasts(overhead, limit)
}
func (d *delegate) LocalState(join bool) []byte {
mtx.RLock()
m := items
mtx.RUnlock()
b, _ := json.Marshal(m)
return b
}
func (d *delegate) MergeRemoteState(buf []byte, join bool) { if len(buf) == 0 {
return
}
大众社会if !join {
return
}
var m map[string]string
if err := json.Unmarshal(buf, &m); err != nil {
return
}
mtx.Lock()
for k, v := range m {
items[k] = v
}
mtx.Unlock()
}
func addHandler(w http.ResponWriter, r *http.Request) { r.ParForm()
key := r.Form.Get("key")
val := r.Form.Get("val")
mtx.Lock()
items[key] = val
mtx.Unlock()
b, err := json.Marshal([]*update{
&update{
Action: "add",
Data: map[string]string{
key: val,
},
},
})
if err != nil {
http.Error(w, err.Error(), 500)
return
}
//⼴播数据
broadcasts.QueueBroadcast(&broadcast{
msg: append([]byte("d"), b...),
notify: nil,
})
}
func delHandler(w http.ResponWriter, r *http.Request) { r.ParForm()
key := r.Form.Get("key")
mtx.Lock()
delete(items, key)
mtx.Unlock()
b, err := json.Marshal([]*update{
梦见亲人死亡
&update{
Action: "del",
Data: map[string]string{
key: "",
},
},
})
if err != nil {
http.Error(w, err.Error(), 500)
return
}
broadcasts.QueueBroadcast(&broadcast{
msg: append([]byte("d"), b...),
叠飞镖notify: nil,
})
}
func getHandler(w http.ResponWriter, r *http.Request) { r.ParForm()
key := r.Form.Get("key")
mtx.RLock()
val := items[key]
mtx.RUnlock()
山东潍坊
w.Write([]byte(val))
}
func start() error {
hostname, _ := os.Hostname()
c := memberlist.DefaultLocalConfig()
c.Delegate = &delegate{}
c.BindPort = 0
c.Name = hostname + "-" + uui
d.NewUUID().String()
//创建gossip⽹络
m, err := memberlist.Create(c)
if err != nil {
return err
}
//第⼀个节点没有member，但从第⼆个开始就有member了
if len(*members) > 0 {
parts := strings.Split(*members, ",")
_, err := m.Join(parts)
if err != nil {
return err
}
}
broadcasts = &memberlist.TransmitLimitedQueue{
NumNodes: func() int {
天不会不蓝
return m.NumMembers()
},
RetransmitMult: 3,
}
node := m.LocalNode()
fmt.Printf("Local member %s:%d\n", node.Addr, node.Port)
return nil
}
桌面图标怎么隐藏
func main() {
if err := start(); err != nil {
fmt.Println(err)
}
http.HandleFunc("/add", addHandler)
http.HandleFunc("/del", delHandler)
http.HandleFunc("/get", getHandler)
fmt.Printf("Listening on :%d\n", *port)
if err := http.ListenAndServe(fmt.Sprintf(":%d", *port), nil); err != nil {
fmt.Println(err)
}
}
这⾥通过⼀个简单的http服务查询和插⼊数据，找两台机器，第⼀台执⾏memberlist
会⽣成gossip监听的服务ip和端⼝
使⽤上⾯的ip和端⼝在第⼆台执⾏
memberlist --x.xxx:xxxx
那么⼀个gossip的⽹络就搭建完成了。
# add
curl "localhost:4001/add?key=foo&val=bar"
# get
curl "另⼀台机器:4001/get?key=foo"
# delete
curl "localhost:4001/del?key=foo"
