~~~ //Nsq接收測試 package main import ( "fmt" "time" "github.com/nsqio/go-nsq" ) // 消費者 type ConsumerT struct{} // 主函數 func main() { InitConsumer("test", "test-channel", "127.0.0.1:4161") for { time.Sleep(time.Second * 10) } } //處理消息 func (*ConsumerT) HandleMessage(msg *nsq.Message) error { fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body)) return nil } //初始化消費者 func InitConsumer(topic string, channel string, address string) { cfg := nsq.NewConfig() cfg.LookupdPollInterval = time.Second //設置重連時間 c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一個消費者 if err != nil { panic(err) } c.SetLogger(nil, 0) //屏蔽系統日志 c.AddHandler(&ConsumerT{}) // 添加消費者接口 //建立NSQLookupd連接 if err := c.ConnectToNSQLookupd(address); err != nil { panic(err) } //建立多個nsqd連接 // if err := c.ConnectToNSQDs([]string{"127.0.0.1:4150", "127.0.0.1:4152"}); err != nil { // panic(err) // } // 建立一個nsqd連接 // if err := c.ConnectToNSQD("127.0.0.1:4150"); err != nil { // panic(err) // } } ~~~