微服務(wù)注冊(cè)中心分布式集群設(shè)計(jì)原理與 Golang 實(shí)現(xiàn)

內(nèi)容提要

服務(wù)注冊(cè)中心設(shè)計(jì)原理與Golang實(shí)現(xiàn)

集群待解決問(wèn)題

節(jié)點(diǎn)關(guān)系與成員發(fā)現(xiàn)

Why You Shouldn’t Use ZooKeeper for Service Discovery

type Node struct {
    config      *configs.Config
    addr        string
    status      int 
}
func NewNode(config *configs.GlobalConfig, addr string) *Node {
    return &Node{
        addr:        addr,
        status:      configs.NodeStatusDown, //初始化設(shè)為下線狀態(tài)
    }   
}

type Nodes struct {
    nodes    []*Node
    selfAddr string
}
//初始化默認(rèn)從配置文件中加載節(jié)點(diǎn)信息
func NewNodes(c *configs.GlobalConfig) *Nodes {
    nodes := make([]*Node, 0, len(c.Nodes))
    for _, addr := range c.Nodes {
        n := NewNode(c, addr)
        nodes = append(nodes, n)
    }
    return &Nodes{
        nodes:    nodes,
        selfAddr: c.HttpServer,
    }   
}

type Discovery struct {
    config    *configs.GlobalConfig
    protected bool
    Registry  *Registry
+   Nodes     atomic.Value
}
func NewDiscovery(config *configs.GlobalConfig) *Discovery {
//...
+ dis.Nodes.Store(NewNodes(config))
}

服務(wù)注冊(cè)中心設(shè)計(jì)原理與Golang實(shí)現(xiàn)

func (dis *Discovery) regSelf() {
    now := time.Now().UnixNano()
    instance := &Instance{
        Env:             dis.config.Env,
        Hostname:        dis.config.Hostname,
        AppId:           configs.DiscoveryAppId, //Kavin.discovery
        Addrs:           []string{"http://" + dis.config.HttpServer},
        Status:          configs.NodeStatusUp,
        RegTimestamp:    now,
        UpTimestamp:     now,
        LatestTimestamp: now,
        RenewTimestamp:  now,
        DirtyTimestamp:  now,
    }
    dis.Registry.Register(instance, now)
    //注冊(cè)后同步到其他集群，下面部分會(huì)展開講解
    dis.Nodes.Load().(*Nodes).Replicate(configs.Register, instance)
}

func (dis *Discovery) renewTask(instance *Instance) {
    now := time.Now().UnixNano()
    ticker := time.NewTicker(configs.RenewInterval) //30 second
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:            
            _, err := dis.Registry.Renew(instance.Env, instance.AppId, instance.Hostname)
            if err == errcode.NotFound {
                dis.Registry.Register(instance, now)
                dis.Nodes.Load().(*Nodes).Replicate(configs.Register, instance)
            } else {
                dis.Nodes.Load().(*Nodes).Replicate(configs.Renew, instance)
            } 
        }   
    }   
}

func main() {
    //graceful restart
    quit := make(chan os.Signal)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT)
    <-quit
    log.Println("shutdown discovery server...")
    //cancel
++  global.Discovery.CancelSelf()
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
}

func (dis *Discovery) nodesPerception() {
    var lastTimestamp int64
    ticker := time.NewTicker(configs.NodePerceptionInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            fetchData, err := dis.Registry.Fetch(dis.config.Env, configs.DiscoveryAppId, configs.NodeStatusUp, lastTimest
amp)
            if err != nil || fetchData == nil {
                continue
            }   
            var nodes []string
            for _, instance := range fetchData.Instances {
                for _, addr := range instance.Addrs {
                    u, err := url.Parse(addr)
                    if err == nil {
                        nodes = append(nodes, u.Host)
                    }   
                }   
            } 
            lastTimestamp = fetchData.LatestTimestamp
            config := new(configs.GlobalConfig)
            *config = *dis.config
            config.Nodes = nodes
            ns := NewNodes(config)
            ns.SetUp()
            dis.Nodes.Store(ns)
        }
    }
}

數(shù)據(jù)副本與數(shù)據(jù)一致性

func (dis *Discovery) initSync() {
    nodes := dis.Nodes.Load().(*Nodes)
    for _, node := range nodes.AllNodes() {
        if node.addr == nodes.selfAddr {
            continue
        }
        uri := fmt.Sprintf("http://%s%s", node.addr, configs.FetchAllURL)
        resp, err := httputil.HttpPost(uri, nil)
        if err != nil {
            log.Println(err)
            continue
        }
        var res struct {
            Code    int                    `json:"code"`
            Message string                 `json:"message"`
            Data    map[string][]*Instance `json:"data"`
        }
        err = json.Unmarshal([]byte(resp), &res)
        if err != nil {
            log.Printf("get from %v error : %v", uri, err)
            continue
        }
        if res.Code != configs.StatusOK {
            log.Printf("get from %v error : %v", uri, res.Message)
            continue
        }
        dis.protected = false
        for _, v := range res.Data {
            for _, instance := range v {
                dis.Registry.Register(instance, instance.LatestTimestamp)
            }
        }
    }
    nodes.SetUp()
}

func RegisterHandler(c *gin.Context) {
   //...
+   if req.Replication {
+       global.Discovery.Nodes.Load().(*model.Nodes).Replicate(c, configs.Register, instance)
+   } 
}

func (nodes *Nodes) Replicate(c *gin.Context, action configs.Action, instance *Instance) error {
    if len(nodes.nodes) == 0 {
        return nil
    }
    for _, node := range nodes.nodes {
        if node.addr != nodes.selfAddr {
            go nodes.action(c, node, action, instance)
        }
    }
    return nil
}
func (nodes *Nodes) action(c *gin.Context, node *Node, action configs.Action, instance *Instance) {
    switch action {
    case configs.Register:
        go node.Register(c, instance)
    case configs.Renew:
        go node.Renew(c, instance)
    case configs.Cancel:
        go node.Cancel(c, instance)
    }
}

func (node *Node) Register(c *gin.Context, instance *Instance) error {
    return node.call(c, node.registerURL, configs.Register, instance, nil)
}
func (node *Node) call(c *gin.Context, uri string, action configs.Action, instance *Instance, data interface{}) error {
    params := make(map[string]interface{})
    params["env"] = instance.Env
    params["appid"] = instance.AppId
    params["hostname"] = instance.Hostname
    params["replication"] = true //broadcast stop here
    switch action {
    case configs.Register:
        params["addrs"] = instance.Addrs
        params["status"] = instance.Status
        params["version"] = instance.Version
        params["reg_timestamp"] = strconv.FormatInt(instance.RegTimestamp, 10) 
        params["dirty_timestamp"] = strconv.FormatInt(instance.DirtyTimestamp, 10) 
        params["latest_timestamp"] = strconv.FormatInt(instance.LatestTimestamp, 10) 
    case configs.Renew:
        params["dirty_timestamp"] = strconv.FormatInt(instance.DirtyTimestamp, 10) 
    case configs.Cancel:
        params["latest_timestamp"] = strconv.FormatInt(instance.LatestTimestamp, 10) 
    }   
    resp, err := httputil.HttpPost(uri, params)
    if err != nil {
        return err 
    }   
    res := Response{}
    err = json.Unmarshal([]byte(resp), &res)
    if err != nil {
        return err 
    }   
    if res.Code != configs.StatusOK {
        json.Unmarshal([]byte(res.Data), data)
        return errcode.Conflict
    }   
    return nil 
}

數(shù)據(jù)分區(qū)策略

總結(jié)