1. 消费者

  1. //Nsq接收测试
  2. package main
  4. import (
  5.     "fmt"
  6.     "time"
  8.     "github.com/nsqio/go-nsq"
  9. )
  11. // 消费者
  12. type ConsumerT struct{}
  14. // 主函数
  15. func main() {
  16.     InitConsumer("test", "test-channel", "127.0.0.1:4161")
  17.     for {
  18.         time.Sleep(time.Second * 10)
  19.     }
  20. }
  22. //处理消息
  23. func (*ConsumerT) HandleMessage(msg *nsq.Message) error {
  24.     fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
  25.     return nil
  26. }
  28. //初始化消费者
  29. func InitConsumer(topic string, channel string, address string) {
  30.     cfg := nsq.NewConfig()
  31.     cfg.LookupdPollInterval = time.Second          //设置重连时间
  32.     c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一个消费者
  33.     if err != nil {
  34.         panic(err)
  35.     }
  36.     c.SetLogger(nil, 0)        //屏蔽系统日志
  37.     c.AddHandler(&ConsumerT{}) // 添加消费者接口
  39.     //建立NSQLookupd连接
  40.     if err := c.ConnectToNSQLookupd(address); err != nil {
  41.         panic(err)
  42.     }
  44.     //建立多个nsqd连接
  45.     // if err := c.ConnectToNSQDs([]string{"127.0.0.1:4150", "127.0.0.1:4152"}); err != nil {
  46.     //  panic(err)
  47.     // }
  49.     // 建立一个nsqd连接
  50.     // if err := c.ConnectToNSQD("127.0.0.1:4150"); err != nil {
  51.     //  panic(err)
  52.     // }
  53. }