> ### 十三例 NSQ操作 * 軟件下載(服務) :[https://nsq.io/deployment/installing.html](https://nsq.io/deployment/installing.html) * * * * nsqlookupd.exe * nsqd.exe --lookupd-tcp-address=127.0.0.1:4160 * nsqadmin.exe --lookupd-http-address localhost:4161 * * * * 一個節點下創建多個消息隊列, 給多個模塊消費  > ### 生產者 ~~~ package main import ( "bufio" "fmt" "os" "strings" "github.com/nsqio/go-nsq" ) func main() { nsqAddress := "127.0.0.1:4150" config := nsq.NewConfig() producer, err := nsq.NewProducer(nsqAddress, config) if err != nil{ fmt.Println(err) } //讀取控制臺輸入 reader := bufio.NewReader(os.Stdin) for { data, err := reader.ReadString('\n') if err != nil { fmt.Printf("read string failed, err:%v\n", err) continue } data = strings.TrimSpace(data) if data == "stop" { break } err = producer.Publish("queue", []byte(data)) if err != nil { fmt.Printf("publish message failed, err:%v\n", err) continue } fmt.Printf("publish data:%s succ\n", data) } } ~~~ > ### 消費者 ~~~ package main import ( "fmt" "os" "syscall" "time" "os/signal" "github.com/nsqio/go-nsq" ) // 消費者 type Consumer struct { } //處理消息 func (*Consumer) HandleMessage(msg *nsq.Message) error { fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body)) return nil } //初始化消費者 func initConsumer(topic string, channel string, address string) error { cfg := nsq.NewConfig() cfg.LookupdPollInterval = 15 * time.Second //設置服務發現的輪詢時間 c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一個消費者 if err != nil { return err } consumer := &Consumer{} c.AddHandler(consumer) // 添加消費者接口 //建立NSQLookupd連接 if err := c.ConnectToNSQLookupd(address); err != nil { return err } return nil } // 主函數 func main() { err := initConsumer("queue", "channel1", "127.0.0.1:4161") if err != nil { fmt.Printf("init consumer failed, err:%v\n", err) return } c := make(chan os.Signal) signal.Notify(c, syscall.SIGINT) <-c } ~~~