nsq

notify/consumer.go

@@ -7,15 +7,20 @@ import (
 )
 
 type NsqConfig struct {
-	Topic   string
-	Channel string
-	Addr    string
-	Handler nsq.Handler
+	Topic       string
+	Channel     string
+	Addr        string
+	MaxInFlight int
+	Handler     nsq.Handler
 }
 
 func InitConsumer(consumer *NsqConfig) {
 	cfg := nsq.NewConfig()
-	cfg.LookupdPollInterval = time.Second * 30                       //设置重连时间
+	cfg.LookupdPollInterval = time.Second * 30
+	if consumer.MaxInFlight > 0 {
+		cfg.MaxInFlight = consumer.MaxInFlight
+	}
+	//设置重连时间
 	c, err := nsq.NewConsumer(consumer.Topic, consumer.Channel, cfg) // 新建一个消费者
 	if logs.CheckErr(err, "InitConsumer") {
 		return
@@ -30,6 +35,9 @@ func InitConsumer(consumer *NsqConfig) {
 
 func InitBatchConsumer(consumer *NsqConfig, clientNum int) {
 	cfg := nsq.NewConfig()
+	if consumer.MaxInFlight > 0 {
+		cfg.MaxInFlight = consumer.MaxInFlight
+	}
 	cfg.LookupdPollInterval = time.Second * 30                       //设置重连时间
 	c, err := nsq.NewConsumer(consumer.Topic, consumer.Channel, cfg) // 新建一个消费者