nsq

notify/consumer.go

@@ -23,22 +23,25 @@ type ConsumerPool struct {
 
 // InitConsumer 初始化单个消费者
 func InitConsumer(consumer *NsqConfig) {
-	cfg := nsq.NewConfig()
-	cfg.LookupdPollInterval = time.Second * 30
+	config := nsq.NewConfig()
+	config.HeartbeatInterval = 5 * time.Second
+	config.MsgTimeout = 30 * time.Second
+	config.DialTimeout = 5 * time.Second
+	config.LookupdPollInterval = time.Second * 30
 	// 优化内存:限制MaxInFlight,避免过多消息堆积在内存中
 	if consumer.MaxInFlight > 0 {
-		cfg.MaxInFlight = consumer.MaxInFlight
+		config.MaxInFlight = consumer.MaxInFlight
 	} else {
-		cfg.MaxInFlight = 1 // 默认最小值,减少内存占用
+		config.MaxInFlight = 1 // 默认最小值,减少内存占用
 	}
 
 	// 设置其他内存优化参数
-	cfg.DefaultRequeueDelay = time.Second * 5
-	cfg.MaxRequeueDelay = time.Minute
-	cfg.ReadTimeout = time.Second * 30
-	cfg.WriteTimeout = time.Second * 10
+	config.DefaultRequeueDelay = time.Second * 5
+	config.MaxRequeueDelay = time.Minute
+	config.ReadTimeout = time.Second * 30
+	config.WriteTimeout = time.Second * 5
 
-	c, err := nsq.NewConsumer(consumer.Topic, consumer.Channel, cfg) // 新建一个消费者
+	c, err := nsq.NewConsumer(consumer.Topic, consumer.Channel, config) // 新建一个消费者
 	if logs.CheckErr(err, "InitConsumer") {
 		return
 	}