Add heatbeat for register

Author: chenzhiguo

api.go

@@ -32,15 +32,21 @@ func Register(zone, app string, instance *Instance) error {
 
 	// status: http.StatusNoContent
 	result := requests.Post(u).Json(info).Send().Status2xx()
+	if result.Err == nil {
+		instance.Beater.AddBeatInfo(app, instance)
+	}
 	return result.Err
 }
 
 // UnRegister 删除实例
 // DELETE /eureka/v2/apps/appID/instanceID
-func UnRegister(zone, app, instanceID string) error {
-	u := zone + "apps/" + app + "/" + instanceID
+func UnRegister(zone, app string, instance *Instance) error {
+	u := zone + "apps/" + app + "/" + instance.InstanceID
 	// status: http.StatusNoContent
 	result := requests.Delete(u).Send().StatusOk()
+	if result.Err == nil && instance.Beater != nil {
+		instance.Beater.RemoveBeatInfo(app, instance.InstanceID)
+	}
 	return result.Err
 }
 

beat.go

@@ -0,0 +1,106 @@
+package eureka_client
+
+import (
+	"context"
+	"log"
+	"sync"
+	"time"
+
+	"golang.org/x/sync/semaphore"
+)
+
+type BeatReactor struct {
+	config              *Config
+	beatMap             ConcurrentMap
+	clientBeatInterval  int64
+	beatThreadCount     int
+	beatThreadSemaphore *semaphore.Weighted
+	beatRecordMap       ConcurrentMap
+	mux                 *sync.Mutex
+	log                 Logger
+	Period              time.Duration
+}
+
+const DefaultBeatThreadNum = 20
+
+var ctx = context.Background()
+
+func NewBeatReactor(config *Config, clientBeatInterval int64) BeatReactor {
+	br := BeatReactor{
+		config: config,
+	}
+	if clientBeatInterval <= 0 {
+		clientBeatInterval = 5 * 1000
+	}
+	br.beatMap = NewConcurrentMap()
+	br.clientBeatInterval = clientBeatInterval
+	br.beatThreadCount = DefaultBeatThreadNum
+	br.beatRecordMap = NewConcurrentMap()
+	br.beatThreadSemaphore = semaphore.NewWeighted(int64(br.beatThreadCount))
+	br.mux = new(sync.Mutex)
+	br.log = NewLogger()
+	br.Period = time.Duration(time.Millisecond.Milliseconds() * clientBeatInterval)
+	return br
+}
+
+func (br *BeatReactor) AddBeatInfo(serviceName string, beatInfo *Instance) {
+	k := beatInfo.InstanceID
+	defer br.mux.Unlock()
+	br.mux.Lock()
+	if data, ok := br.beatMap.Get(k); ok {
+		beatInfo = data.(*Instance)
+		beatInfo.Status = "UP"
+		br.beatMap.Remove(k)
+	}
+	br.beatMap.Set(k, beatInfo)
+	go br.sendInstanceBeat(k, beatInfo)
+}
+
+func (br *BeatReactor) RemoveBeatInfo(serviceName string, instanceId string) {
+	log.Printf("remove beat: %s@%s from beat map", serviceName, instanceId)
+	k := instanceId
+	defer br.mux.Unlock()
+	br.mux.Lock()
+	data, exist := br.beatMap.Get(k)
+	if exist {
+		beatInfo := data.(*Instance)
+		beatInfo.Status = "UP"
+	}
+	br.beatMap.Remove(k)
+}
+
+func (br *BeatReactor) sendInstanceBeat(k string, beatInfo *Instance) {
+	for {
+		err := br.beatThreadSemaphore.Acquire(ctx, 1)
+		if err != nil {
+			log.Printf("sendInstanceBeat failed to acquire semaphore: %v", err)
+			return
+		}
+		//如果当前实例注销，则进行停止心跳
+		if beatInfo.Status != "UP" {
+			log.Printf("instance[%s] stop heartBeating", k)
+			br.beatThreadSemaphore.Release(1)
+			return
+		}
+
+		//进行心跳通信
+		// /eureka/apps/ORDER-SERVICE/localhost:order-service:8886?status=UP
+		err = Heartbeat(br.config.DefaultZone, beatInfo.App, beatInfo.InstanceID)
+		//u := br.config.DefaultZone + "apps/" + beatInfo.App + "/" + beatInfo.InstanceID + "?status=UP"
+		//result := requests.Put(u).Send().Status2xx()
+
+		if err != nil {
+			log.Printf("beat to server return error:%+v", err)
+			br.beatThreadSemaphore.Release(1)
+			t := time.NewTimer(br.Period)
+			<-t.C
+			continue
+		}
+
+		br.beatRecordMap.Set(k, time.Now().UnixNano()/1e6)
+		br.beatThreadSemaphore.Release(1)
+
+		t := time.NewTimer(br.Period)
+		<-t.C
+	}
+}

client.go

@@ -50,7 +50,7 @@ func (c *Client) Start() {
 // refresh 刷新服务列表
 func (c *Client) refresh() {
 	timer := time.NewTimer(0)
-	interval := time.Duration(c.Config.RenewalIntervalInSecs) * time.Second
+	interval := time.Duration(c.Config.RegistryFetchIntervalSeconds) * time.Second
 	for c.running {
 		select {
 		case <-timer.C:
@@ -70,7 +70,7 @@ func (c *Client) refresh() {
 // heartbeat 心跳
 func (c *Client) heartbeat() {
 	timer := time.NewTimer(0)
-	interval := time.Duration(c.Config.RegistryFetchIntervalSeconds) * time.Second
+	interval := time.Duration(c.Config.RenewalIntervalInSecs) * time.Second
 	for c.running {
 		select {
 		case <-timer.C: