nsq

notify/consumer.go

@@ -11,6 +11,7 @@ type NsqConfig struct {
 	Channel     string
 	Addr        string
 	MaxInFlight int
+	Concurrency int
 	Handler     nsq.Handler
 }
 
@@ -25,7 +26,11 @@ func InitConsumer(consumer *NsqConfig) {
 	if logs.CheckErr(err, "InitConsumer") {
 		return
 	}
+	if consumer.Concurrency > 1 {
+		c.AddConcurrentHandlers(consumer.Handler, consumer.Concurrency)
+	} else {
 		c.AddHandler(consumer.Handler) // 添加消费者接口
+	}
 	//建立NSQLookupd连接
 	err = c.ConnectToNSQD(consumer.Addr)
 	if logs.CheckErr(err, "ConnectToNSQD") {