聊聊nacossdkgo的HostReactor

序

本文主要研究一下 nacos-sdk-go 的 HostReactor

HostReactor

nacos-sdk-go-v0.3.2/clients/naming_client/host_reator.go

type HostReactor struct {
    serviceInfoMap       cache.ConcurrentMap
    cacheDir             string
    updateThreadNum      int
    serviceProxy         NamingProxy
    pushReceiver         PushReceiver
    subCallback          SubscribeCallback
    updateTimeMap        cache.ConcurrentMap
    updateCacheWhenEmpty bool
}

• HostReactor 定义了 serviceInfoMap、cacheDir、updateThreadNum、serviceProxy、pushReceiver、subCallback、updateTimeMap、updateCacheWhenEmpty 属性

NewHostReactor

nacos-sdk-go-v0.3.2/clients/naming_client/host_reator.go

func NewHostReactor(serviceProxy NamingProxy, cacheDir string, updateThreadNum int, notLoadCacheAtStart bool, subCallback SubscribeCallback, updateCacheWhenEmpty bool) HostReactor {
    if updateThreadNum <= 0 {updateThreadNum = Default_Update_Thread_Num}
    hr := HostReactor{
        serviceProxy:         serviceProxy,
        cacheDir:             cacheDir,
        updateThreadNum:      updateThreadNum,
        serviceInfoMap:       cache.NewConcurrentMap(),
        subCallback:          subCallback,
        updateTimeMap:        cache.NewConcurrentMap(),
        updateCacheWhenEmpty: updateCacheWhenEmpty,
    }
    pr := NewPushRecevier(&hr)
    hr.pushReceiver = *pr
    if !notLoadCacheAtStart {hr.loadCacheFromDisk()
    }
    go hr.asyncUpdateService()
    return hr
}

• NewHostReactor 方法创建 HostReactor，然后通过 NewPushRecevier 创建 pushReceiver，对于 notLoadCacheAtStart 为 false 的则执行 loadCacheFromDisk，之后异步执行 asyncUpdateService

loadCacheFromDisk

nacos-sdk-go-v0.3.2/clients/naming_client/host_reator.go

func (hr *HostReactor) loadCacheFromDisk() {serviceMap := cache.ReadServicesFromFile(hr.cacheDir)
    if serviceMap == nil || len(serviceMap) == 0 {return}
    for k, v := range serviceMap {hr.serviceInfoMap.Set(k, v)
    }
}

• loadCacheFromDisk 方法通过 cache.ReadServicesFromFile(hr.cacheDir) 获取 serviceMap

asyncUpdateService

nacos-sdk-go-v0.3.2/clients/naming_client/host_reator.go

func (hr *HostReactor) asyncUpdateService() {sema := utils.NewSemaphore(hr.updateThreadNum)
    for {for _, v := range hr.serviceInfoMap.Items() {service := v.(model.Service)
            lastRefTime, ok := hr.updateTimeMap.Get(utils.GetServiceCacheKey(service.Name, service.Clusters))
            if !ok {lastRefTime = uint64(0)
            }
            if uint64(utils.CurrentMillis())-lastRefTime.(uint64) > service.CacheMillis {sema.Acquire()
                go func() {hr.updateServiceNow(service.Name, service.Clusters)
                    sema.Release()}()}
        }
        time.Sleep(1 * time.Second)
    }

}

• asyncUpdateService 方法遍历 serviceInfoMap，通过 hr.updateTimeMap.Get(utils.GetServiceCacheKey(service.Name, service.Clusters)) 获取 lastRefTime，然后判断是否超过 service.CacheMillis，超过的 haul 则执行 sema.Acquire()，异步 hr.updateServiceNow(service.Name, service.Clusters)，最后执行 sema.Release()

updateServiceNow

nacos-sdk-go-v0.3.2/clients/naming_client/host_reator.go

func (hr *HostReactor) updateServiceNow(serviceName string, clusters string) {result, err := hr.serviceProxy.QueryList(serviceName, clusters, hr.pushReceiver.port, false)
    if err != nil {log.Printf("[ERROR]:query list return error!servieName:%s cluster:%s  err:%s \n", serviceName, clusters, err.Error())
        return
    }
    if result == "" {log.Printf("[ERROR]:query list is empty!servieName:%s cluster:%s \n", serviceName, clusters)
        return
    }
    hr.ProcessServiceJson(result)
}

• updateServiceNow 方法通过 hr.serviceProxy.QueryList(serviceName, clusters, hr.pushReceiver.port, false) 获取 json，然后通过 hr.ProcessServiceJson(result) 解析 json

ProcessServiceJson

nacos-sdk-go-v0.3.2/clients/naming_client/host_reator.go

func (hr *HostReactor) ProcessServiceJson(result string) {service := utils.JsonToService(result)
    if service == nil {return}
    cacheKey := utils.GetServiceCacheKey(service.Name, service.Clusters)

    oldDomain, ok := hr.serviceInfoMap.Get(cacheKey)
    if ok && !hr.updateCacheWhenEmpty {
        //if instance list is empty,not to update cache
        if service.Hosts == nil || len(service.Hosts) == 0 {log.Printf("[ERROR]:do not have useful host, ignore it, name:%s \n", service.Name)
            return
        }
    }
    hr.updateTimeMap.Set(cacheKey, uint64(utils.CurrentMillis()))
    hr.serviceInfoMap.Set(cacheKey, *service)
    if !ok || ok && !reflect.DeepEqual(service.Hosts, oldDomain.(model.Service).Hosts) {
        if !ok {log.Println("[INFO] service not found in cache" + cacheKey)
        } else {log.Printf("[INFO] service key:%s was updated to:%s \n", cacheKey, utils.ToJsonString(service))
        }
        cache.WriteServicesToFile(*service, hr.cacheDir)
        hr.subCallback.ServiceChanged(service)
    }
}

• ProcessServiceJson 方法通过 utils.JsonToService(result) 将 json 解析为 model.Service，然后通过 utils.GetServiceCacheKey(service.Name, service.Clusters) 构建 cacheKey，之后更新 hr.updateTimeMap、hr.serviceInfoMap；对于缓存不存在或者缓存存在变更的则执行 cache.WriteServicesToFile(*service, hr.cacheDir)，然后触发 hr.subCallback.ServiceChanged(service)

小结

HostReactor 定义了 serviceInfoMap、cacheDir、updateThreadNum、serviceProxy、pushReceiver、subCallback、updateTimeMap、updateCacheWhenEmpty 属性

doc

• host_reator