更新配置

@@ -78,6 +78,8 @@ spec:
               value: "6DwjBO735"
             - name: BUSINESS_ADMIN_HOST
               value: "http://suplus-business-admin-dev.fjmaimaimai.com"
+            - name: KAFKA_HOST
+              value: "106.52.15.41:9092"
       volumes:
         - name: accesslogs
           emptyDir: {}

@@ -75,6 +75,8 @@ spec:
               value: "rsF0pL!6DwjBO735"
             - name: BUSINESS_ADMIN_HOST
               value: "http://suplus-business-admin-test.fjmaimaimai.com"
+            - name: KAFKA_HOST
+              value: "106.52.15.41:9092"
       volumes:
         - name: accesslogs
           emptyDir: {}

+package constant
+
+import (
+	"os"
+	"strings"
+)
+
+type KafkaConfig struct {
+	Servers    []string `json:"servers"`
+	ConsumerId string   `json:"consumerGroup"`
+}
+
+var KafkaCfg KafkaConfig
+
+func init() {
+	KafkaCfg = KafkaConfig{
+		Servers:    []string{"106.52.15.41:9092"},
+		ConsumerId: "partnermg",
+	}
+	if os.Getenv("KAFKA_HOST") != "" {
+		kafkaHost := os.Getenv("KAFKA_HOST")
+		KafkaCfg.Servers = strings.Split(kafkaHost, ";")
+	}
+}

@@ -3,6 +3,8 @@ package configs
 import (
 	"os"
 	"strings"
+
+	"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/constant"
 )
 
 type MqConfig struct {
@@ -10,7 +12,10 @@ type MqConfig struct {
 	ConsumerId string   `json:"consumerGroup"`
 }
 
-var Cfg MqConfig
+var Cfg = MqConfig{
+	Servers:    constant.KafkaCfg.Servers,
+	ConsumerId: constant.KafkaCfg.ConsumerId,
+}
 
 func init() {
 

@@ -59,8 +59,9 @@ func (c *MessageConsumer) ConsumeClaim(groupSession sarama.ConsumerGroupSession,
 		}
 		if err = topicHandle(message); err != nil {
 			logs.Error("Message claimed: kafka消息处理错误 topic =", message.Topic, message.Offset, err)
+		} else {
+			groupSession.MarkMessage(message, "")
 		}
-		groupSession.MarkMessage(message, "")
 	}
 	return nil
 }