聊聊nacossdkgo的BeatReactor

序

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

BeatReactor

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

type BeatReactor struct {
    beatMap             cache.ConcurrentMap
    serviceProxy        NamingProxy
    clientBeatInterval  int64
    beatThreadCount     int
    beatThreadSemaphore *nsema.Semaphore
    beatRecordMap       cache.ConcurrentMap
}

• BeatReactor 定义了 beatMap、serviceProxy、clientBeatInterval、beatThreadCount、beatThreadSemaphore、beatRecordMap 属性

NewBeatReactor

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

func NewBeatReactor(serviceProxy NamingProxy, clientBeatInterval int64) BeatReactor {br := BeatReactor{}
    if clientBeatInterval <= 0 {clientBeatInterval = 5 * 1000}
    br.beatMap = cache.NewConcurrentMap()
    br.serviceProxy = serviceProxy
    br.clientBeatInterval = clientBeatInterval
    br.beatThreadCount = Default_Beat_Thread_Num
    br.beatRecordMap = cache.NewConcurrentMap()
    br.beatThreadSemaphore = nsema.NewSemaphore(br.beatThreadCount)
    return br
}

• NewBeatReactor 方法创建了 BeatReactor，并初始化其属性

AddBeatInfo

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

func (br *BeatReactor) AddBeatInfo(serviceName string, beatInfo model.BeatInfo) {log.Printf("[INFO] adding beat: <%s> to beat map.\n", utils.ToJsonString(beatInfo))
    k := buildKey(serviceName, beatInfo.Ip, beatInfo.Port)
    br.beatMap.Set(k, &beatInfo)
    go br.sendInstanceBeat(k, &beatInfo)
}

• AddBeatInfo 方法通过 buildKey 构建 key，然后将其放进去 beatMap，之后异步执行 br.sendInstanceBeat

RemoveBeatInfo

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

func (br *BeatReactor) RemoveBeatInfo(serviceName string, ip string, port uint64) {log.Printf("[INFO] remove beat: %s@%s:%d from beat map.\n", serviceName, ip, port)
    k := buildKey(serviceName, ip, port)
    data, exist := br.beatMap.Get(k)
    if exist {beatInfo := data.(*model.BeatInfo)
        beatInfo.Stopped = true
    }
    br.beatMap.Remove(k)
}

• RemoveBeatInfo 方法将指定 key 从 beatMap 移除，对于已经在 beatMap 中的设置其 Stopped 为 true

sendInstanceBeat

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

func (br *BeatReactor) sendInstanceBeat(k string, beatInfo *model.BeatInfo) {
    for {br.beatThreadSemaphore.Acquire()
        // 如果当前实例注销，则进行停止心跳
        if beatInfo.Stopped {log.Printf("[INFO] intance[%s] stop heartBeating\n", k)
            br.beatThreadSemaphore.Release()
            return
        }

        // 进行心跳通信
        beatInterval, err := br.serviceProxy.SendBeat(*beatInfo)
        if err != nil {log.Printf("[ERROR]:beat to server return error:%s \n", err.Error())
            br.beatThreadSemaphore.Release()
            t := time.NewTimer(beatInfo.Period)
            <-t.C
            continue
        }
        if beatInterval > 0 {beatInfo.Period = time.Duration(time.Millisecond.Nanoseconds() * beatInterval)
        }

        br.beatRecordMap.Set(k, utils.CurrentMillis())
        br.beatThreadSemaphore.Release()

        t := time.NewTimer(beatInfo.Period)
        <-t.C
    }
}

• sendInstanceBeat 方法先执行 br.beatThreadSemaphore.Acquire()，之后通过 br.serviceProxy.SendBeat(*beatInfo) 来发送 beat，之后执行 br.beatThreadSemaphore.Release()，然后新创建 time.NewTimer(beatInfo.Period)

小结

BeatReactor 定义了 beatMap、serviceProxy、clientBeatInterval、beatThreadCount、beatThreadSemaphore、beatRecordMap 属性；它提供了 NewBeatReactor、AddBeatInfo、RemoveBeatInfo 方法

doc

• beat_reactor