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

內(nèi)容提要

為什么引入服務(wù)注冊(cè)發(fā)現(xiàn)

注冊(cè)中心實(shí)現(xiàn)原理

type Registry struct {
    apps map[string]*Application
    lock sync.RWMutex
}

type Application struct {
    appid           string
    instances       map[string]*Instance
    latestTimestamp int64
    lock            sync.RWMutex
}

type Instance struct {
    Env      string   `json:"env"`
    AppId    string   `json:"appid"`
    Hostname string   `json:"hostname"`
    Addrs    []string `json:"addrs"`
    Version  string   `json:"version"`
    Status   uint32   `json:"status"`

    RegTimestamp    int64 `json:"reg_timestamp"`
    UpTimestamp     int64 `json:"up_timestamp"`
    RenewTimestamp  int64 `json:"renew_timestamp"`
    DirtyTimestamp  int64 `json:"dirty_timestamp"`
    LatestTimestamp int64 `json:"latest_timestamp"`
}

func NewRegistry() *Registry {
    registry := &Registry{
        apps: make(map[string]*Application),
    }
    return registry
}
func NewInstance(req *RequestRegister) *Instance {
    now := time.Now().UnixNano()
    instance := &Instance{
        Env:             req.Env,
        AppId:           req.AppId,
        Hostname:        req.Hostname,
        Addrs:           req.Addrs,
        Version:         req.Version,
        Status:          req.Status,
        RegTimestamp:    now,
        UpTimestamp:     now,
        RenewTimestamp:  now,
        DirtyTimestamp:  now,
        LatestTimestamp: now,
    }
    return instance
}
r := NewRegistry()
instance := NewInstance(&req)
r.Register(instance, req.LatestTimestamp)

func (r *Registry) Register(instance *Instance, latestTimestamp int64) (*Application, *errcode.Error) {
    key := getKey(instance.AppId, instance.Env)
    r.lock.RLock()
    app, ok := r.apps[key]
    r.lock.RUnlock()
    if !ok { //new app
        app = NewApplication(instance.AppId)
    }
    //add instance
    _, isNew := app.AddInstance(instance, latestTimestamp)
    if isNew { //todo }
    //add into registry apps
    r.lock.Lock()
    r.apps[key] = app
    r.lock.Unlock()
    return app, nil
}

func NewApplication(appid string) *Application {
    return &Application{
        appid:     appid,
        instances: make(map[string]*Instance),
    }
}   

func (app *Application) AddInstance(in *Instance, latestTimestamp int64) (*Instance, bool) {
    app.lock.Lock() 
    defer app.lock.Unlock()
    appIns, ok := app.instances[in.Hostname]
    if ok { //exist
        in.UpTimestamp = appIns.UpTimestamp
        //dirtytimestamp
        if in.DirtyTimestamp < appIns.DirtyTimestamp {
            log.Println("register exist dirty timestamp")
            in = appIns
        }
    }
    //add or update instances
    app.instances[in.Hostname] = in
    app.upLatestTimestamp(latestTimestamp)
    returnIns := new(Instance)
    *returnIns = *in
    return returnIns, !ok
}   

var req = &model.RequestRegister{AppId: "com.xx.testapp", Hostname: "myhost", Addrs: []string{"http://testapp.xx.com/myhost"}, Status: 1}
func TestRegister(t *testing.T) {
    r := model.NewRegistry()
    instance := model.NewInstance(req)
    app, _ := r.Register(instance, req.LatestTimestamp)
    t.Log(app)
}

r := model.NewRegistry()
fetchData, err := r.Fetch(req.Env, req.AppId, req.Status, 0)

func (r *Registry) Fetch(env, appid string, status uint32, latestTime int64) (*FetchData, *errcode.Error) {
    app, ok := r.getApplication(appid, env)
    if !ok {
        return nil, errcode.NotFound
    }
    return app.GetInstance(status, latestTime)
}
func (r *Registry) getApplication(appid, env string) (*Application, bool) {
    key := getKey(appid, env)
    r.lock.RLock() 
    app, ok := r.apps[key]
    r.lock.RUnlock()
    return app, ok
}

func (app *Application) GetInstance(status uint32, latestTime int64) (*FetchData, *errcode.Error) {
    app.lock.RLock()
    defer app.lock.RUnlock()
    if latestTime >= app.latestTimestamp {
        return nil, errcode.NotModified
    }
    fetchData := FetchData{
        Instances:       make([]*Instance, 0),
        LatestTimestamp: app.latestTimestamp,
    }
    var exists bool
    for _, instance := range app.instances {
        if status&instance.Status > 0 {
            exists = true
            newInstance := copyInstance(instance)
            fetchData.Instances = append(fetchData.Instances, newInstance)
        }
    }
    if !exists {
        return nil, errcode.NotFound
    }
    return &fetchData, nil
}
//deep copy
func copyInstance(src *Instance) *Instance {
    dst := new(Instance)
    *dst = *src
    //copy addrs
    dst.Addrs = make([]string, len(src.Addrs))
    for i, addr := range src.Addrs {
        dst.Addrs[i] = addr
    }
    return dst
}

r := model.NewRegistry()
r.Cancel(req.Env, req.AppId, req.Hostname, 0)

func (r *Registry) Cancel(env, appid, hostname string, latestTimestamp int64) (*Instance, *errcode.Error) {
    log.Println("action cancel...")
    //find app
    app, ok := r.getApplication(appid, env)
    if !ok {
        return nil, errcode.NotFound
    }   
    instance, ok, insLen := app.Cancel(hostname, latestTimestamp)
    if !ok {
        return nil, errcode.NotFound
    }   
    //if instances is empty, delete app from apps
    if insLen == 0 { 
        r.lock.Lock()
        delete(r.apps, getKey(appid, env))
        r.lock.Unlock()
    }   
    return instance, nil 
}
func (app *Application) Cancel(hostname string, latestTimestamp int64) (*Instance, bool, int) {
    newInstance := new(Instance)
    app.lock.Lock()
    defer app.lock.Unlock()
    appIn, ok := app.instances[hostname]
    if !ok {
        return nil, ok, 0
    }   
    //delete hostname
    delete(app.instances, hostname)
    appIn.LatestTimestamp = latestTimestamp
    app.upLatestTimestamp(latestTimestamp)
    *newInstance = *appIn
    return newInstance, true, len(app.instances)
}   

r := model.NewRegistry()
r.Renew(req.Env, req.AppId, req.Hostname)

func (r *Registry) Renew(env, appid, hostname string) (*Instance, *errcode.Error) {
    app, ok := r.getApplication(appid, env)
    if !ok {
        return nil, errcode.NotFound
    }
    in, ok := app.Renew(hostname)
    if !ok {
        return nil, errcode.NotFound
    }       
    return in, nil
}  
func (app *Application) Renew(hostname string) (*Instance, bool) {
    app.lock.Lock()
    defer app.lock.Unlock()
    appIn, ok := app.instances[hostname]
    if !ok {
        return nil, ok
    }
    appIn.RenewTimestamp = time.Now().UnixNano()
    return copyInstance(appIn), true
} 

func NewRegistry() *Registry {
++ go r.evictTask()
}

func (r *Registry) evictTask() {
    ticker := time.Tick(configs.CheckEvictInterval)
    for {
        select {
        case <-ticker:
            r.evict()
        }
    }
}

func (r *Registry) evict() {
    now := time.Now().UnixNano()
    var expiredInstances []*Instance
    apps := r.getAllApplications()
    var registryLen int
    for _, app := range apps {
        registryLen += app.GetInstanceLen()
        allInstances := app.GetAllInstances()
        for _, instance := range allInstances {
            if now-instance.RenewTimestamp > int64(configs.InstanceExpireDuration) {
                expiredInstances = append(expiredInstances, instance)
            }
        }
    }
    evictionLimit := registryLen - int(float64(registryLen)*configs.SelfProtectThreshold)
    expiredLen := len(expiredInstances)
    if expiredLen > evictionLimit {
        expiredLen = evictionLimit
    }

    if expiredLen == 0 {
        return
    }
    for i := 0; i < expiredLen; i++ {
        j := i + rand.Intn(len(expiredInstances)-i)
        expiredInstances[i], expiredInstances[j] = expiredInstances[j], expiredInstances[i]
        expiredInstance := expiredInstances[i]
        r.Cancel(expiredInstance.Env, expiredInstance.AppId, expiredInstance.Hostname, now)
    }
}

type Guard struct {
    renewCount     int64
    lastRenewCount int64
    needRenewCount int64
    threshold      int64
    lock           sync.RWMutex
}

func (gd *Guard) incrNeed() {
    gd.lock.Lock()
    defer gd.lock.Unlock()
    gd.needRenewCount += int64(configs.CheckEvictInterval / configs.RenewInterval)
    gd.threshold = int64(float64(gd.needRenewCount) * configs.SelfProtectThreshold)
}
func (gd *Guard) decrNeed() {
    gd.lock.Lock()
    defer gd.lock.Unlock()
    gd.needRenewCount -= int64(configs.CheckEvictInterval / configs.RenewInterval)
    gd.threshold = int64(float64(gd.needRenewCount) * configs.SelfProtectThreshold)
}
func (gd *Guard) setNeed(count int64) {
    gd.lock.Lock()
    defer gd.lock.Unlock()
    gd.needRenewCount = count * int64(configs.CheckEvictInterval/configs.RenewInterval)
    gd.threshold = int64(float64(gd.needRenewCount) * configs.SelfProtectThreshold)
}
func (gd *Guard) incrCount() {
    atomic.AddInt64(&gd.renewCount, 1)
}

type Registry struct {
++    gd   *Guard
}
func NewRegistry() *Registry {
    r := &Registry{
++      gd:   new(Guard),
    } 
}
func (r *Registry) Register(...) {
    if isNew {
++      r.gd.incrNeed()
    }
}
func (r *Registry) Cancel(...) {
++   r.gd.decrNeed()
}
func (r *Registry) Renew(...) {
++   r.gd.incrCount()
}

func (gd *Guard) storeLastCount() {
    atomic.StoreInt64(&gd.lastRenewCount, atomic.SwapInt64(&gd.needRenewCount, 0))
}
func (gd *Guard) selfProtectStatus() bool {
    return atomic.LoadInt64(&gd.lastRenewCount) < atomic.LoadInt64(&gd.threshold)
}

func (r *Registry) evictTask() {
        case <-ticker:
++          r.gd.storeLastCount()
            r.evict()
        }
}
func (r *Registry) evict() {
   delta := now - instance.RenewTimestamp
++ if !protectStatus && delta > int64(configs.InstanceExpireDuration) ||
      delta > int64(configs.InstanceMaxExpireDuration) {
      expiredInstances = append(expiredInstances, instance)
   }
}

func (r *Registry) evictTask() {
    resetTicker := time.Tick(configs.ResetGuardNeedCountInterval)
    for {
        select { 
        case <-resetTicker:
            var count int64
            for _, app := range r.getAllApplications() {
                count += int64(app.GetInstanceLen())
            }
            r.gd.setNeed(count)
        }
    }
}   

func main() {
    //init config
    c := flag.String("c", "", "config file path")
    flag.Parse()
    config, err := configs.LoadConfig(*c)
    if err != nil {
        log.Println("load config error:", err)
        return
    }  
    //global discovery
    global.Discovery = model.NewDiscovery(config)
    //init router and start server
    router := api.InitRouter()
    srv := &http.Server{
        Addr:    config.HttpServer,
        Handler: router,
    }
    go func() {
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("listen:%s\n", err)
        }
    }()
}

type Discovery struct {
    config    *configs.GlobalConfig
    protected bool
    Registry  *Registry
}
func NewDiscovery(config *configs.GlobalConfig) *Discovery {
    dis := &Discovery{
        protected: false,
        config:    config,
        Registry:  NewRegistry(), //init registry
    }  
    return dis 
}
//init discovery
var Discovery *model.Discovery

router.POST("api/register", handler.RegisterHandler)
func RegisterHandler(c *gin.Context) {
    var req model.RequestRegister
    if e := c.ShouldBindJSON(&req); e != nil {
        err := errcode.ParamError
        c.JSON(http.StatusOK, gin.H{
            "code":    err.Code(),
            "message": err.Error(),
        })
        return
    }
    //bind instance
    instance := model.NewInstance(&req)
    if instance.Status == 0 || instance.Status > 2 {
        err := errcode.ParamError
        c.JSON(http.StatusOK, gin.H{
            "code":    err.Code(),
            "message": err.Error(),
        })
        return
    }
    //dirtytime
    if req.DirtyTimestamp > 0 {
        instance.DirtyTimestamp = req.DirtyTimestamp
    }
    global.Discovery.Registry.Register(instance, req.LatestTimestamp)
    c.JSON(http.StatusOK, gin.H{
        "code":    200,
        "message": "",
        "data":    "",
    })
}

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...")
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    if err := srv.Shutdown(ctx); err != nil {
        log.Fatal("server shutdown error:", err)
    }
    select {
    case <-ctx.Done():
        log.Println("timeout of 5 seconds")
    }
    log.Println("server exiting")
}

總結(jié)與問(wèn)題