Merge pull request #3852 from wcy00000000000000/v3.2.x-datacollection

修复3.2 datacollection空指针异常和连接未关闭的问题

Author: breezeli

src/scene_server/datacollection/datacollection/discover.go

@@ -214,12 +214,14 @@ func (d *Discover) subChan() {
 	if nil != err {
 		d.interrupt <- err
 		blog.Errorf("subscribe [%s] failed: %s", d.chanName, err.Error())
+		return
 	}
 
 	defer func() {
 		subChan.Unsubscribe(d.chanName)
 		d.isSubing = false
 		blog.Infof("close subscribe channel: %s", d.chanName)
+		_ = subChan.Close()
 	}()
 
 	var ts = time.Now()
@@ -244,7 +246,7 @@ func (d *Discover) subChan() {
 
 			blog.Warnf("receive message err: %s", err.Error())
 			d.interrupt <- err
-			continue
+			return
 		}
 
 		msg, ok := received.(*redis.Message)

src/scene_server/datacollection/datacollection/hostsnap.go

@@ -533,8 +533,10 @@ func (h *HostSnap) subChan(snapcli *redis.Client, chanName []string) {
 	subChan, err := snapcli.Subscribe(chanName...)
 	if nil != err {
 		h.interrupt <- err
-		blog.Error("subscribe channel faile ", err.Error())
+		blog.Errorf("subscribe channel failed %s", err.Error())
+		return
 	}
+	defer subChan.Close()
 	closeChan := make(chan struct{})
 	go h.healthCheck(closeChan)
 	defer func() {
@@ -559,13 +561,13 @@ func (h *HostSnap) subChan(snapcli *redis.Client, chanName []string) {
 		if err == redis.Nil || err == io.EOF {
 			continue
 		}
-		msg, ok := received.(*redis.Message)
-		if !ok {
-			continue
-		}
 		if nil != err {
-			blog.Debug("receive messave  err", err.Error())
+			blog.Debug("receive message err", err.Error())
 			h.interrupt <- err
+			return
+		}
+		msg, ok := received.(*redis.Message)
+		if !ok {
 			continue
 		}