调整consumer结构

service/consumer/health_message_handler.go

@@ -22,7 +22,7 @@ func (HealthMessageHandler) MsgProcess(msg string) {
 	err := json.Unmarshal([]byte(msg), &chunkMsg)
 	if err != nil {
 		logger.Error.Println("healthMessageHandler解析json失败:",err)
-		logger.Error.Println([]byte(msg))
+		logger.Error.Println(msg)
 	}
 	buildMsg(chunkMsg)
 }

service/consumer/kafka_agent_sarama.go

@@ -7,11 +7,14 @@ import (
 	"github.com/bsm/sarama-cluster"
 	"os"
 	"os/signal"
+	"sync/atomic"
+	"syscall"
 )
 
+var consumerCount int32
+
 func AgentClusterConsumer(kafkaConf conf.KafkaConf, messageHandle MessageHandler) {
 	config := cluster.NewConfig()
-
 	config.Consumer.Return.Errors = true
 	config.Consumer.Offsets.Initial = sarama.OffsetOldest
 	config.Group.Return.Notifications = true
@@ -20,12 +23,18 @@ func AgentClusterConsumer(kafkaConf conf.KafkaConf, messageHandle MessageHandler
 	if err != nil {
 		panic(err)
 	}
+	atomic.AddInt32(&consumerCount, 1)
 	// trap SIGINT to trigger a shutdown.
 	signals := make(chan os.Signal, 1)
-	signal.Notify(signals, os.Interrupt)
+	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT, os.Interrupt)
 
 	defer func() {
 		_ = consumer.Close()
+		atomic.AddInt32(&consumerCount, -1)
+		logger.Info.Println("consumer结束")
+		if consumerCount == 0 {
+			os.Exit(0)
+		}
 	}()
 
 	// consume errors
@@ -55,5 +64,4 @@ func AgentClusterConsumer(kafkaConf conf.KafkaConf, messageHandle MessageHandler
 			return
 		}
 	}
-
 }