Golang使用NSQ示例

港控/mmm° 2022-05-12 11:56 319阅读 0赞

一、下载

安装包下载地址：[https://nsq.io/deployment/installing.html][https_nsq.io_deployment_installing.html]

nsq源码地址：[https://github.com/nsqio/nsq][https_github.com_nsqio_nsq]

官方提供的golang版本客户端：[https://github.com/nsqio/go-nsq][https_github.com_nsqio_go-nsq]

二、运行NSQ

*   首先启动nsqlookud

nsqlookupd

*  启动nsqd，并接入刚刚启动的nsqlookud。这里为了方便接下来的测试，启动了两个nsqd

nsqd --lookupd-tcp-address=127.0.0.1:4160

nsqd --lookupd-tcp-address=127.0.0.1:4160 --tcp-address=0.0.0.0:4152 --http-address=0.0.0.0:4153

*  启动nqsadmin

nsqadmin --lookupd-http-address=127.0.0.1:4161

三、示例

使用官方提供的golang客户端：

go get github.com/nsqio/go-nsq

1.生成者

ps：开启两个生产者producer1和producer2，由这两个生产者交替生产消息

package main
    
    import (
    	"fmt"
    	"log"
    	"bufio"
    	"os"
    
    	"github.com/nsqio/go-nsq"
    )
    
    
    func main() {
    	strIP1 := "127.0.0.1:4150"
    	strIP2 := "127.0.0.1:4152"
    
    	producer1,err := initProducer(strIP1)
    	if err != nil {
    		log.Fatal("init producer1 error:",err)
    	}
    	producer2,err := initProducer(strIP2)
    	if err != nil {
    		log.Fatal("init producer2 error:",err)
    	}
    
    	defer producer1.Stop()
    	defer producer2.Stop()
    
    	//读取控制台输入
    	reader := bufio.NewReader(os.Stdin)
    
    	count := 0
    	for {
    		fmt.Print("please say:")
    		data, _, _ := reader.ReadLine()
    		command := string(data)
    		if command == "stop" {
    			fmt.Println("stop producer!")
    			return
    		}
    		if count % 2 == 0 {
    			err := producer1.public("test1",command)
    			if err != nil {
    				log.Fatal("producer1 public error:",err)
    			}
    		}else {
    			err := producer2.public("test2",command)
    			if err != nil {
    				log.Fatal("producer2 public error:",err)
    			}
    		}
    
    		count++
    	}
    }
    
    type nsqProducer struct {
    	 *nsq.Producer
    }
    
    
    //初始化生产者
    func initProducer(addr string) (*nsqProducer, error) {
    	fmt.Println("init producer address:",addr)
    	producer,err := nsq.NewProducer(addr,nsq.NewConfig())
    	if err != nil {
    		return nil,err
    	}
    	return &nsqProducer{producer},nil
    }
    
    //发布消息
    func (np *nsqProducer)public(topic,message string) error {
    	err := np.Publish(topic,[]byte(message))
    	if err != nil {
    		log.Println("nsq public error:",err)
    		return err
    	}
    	return nil
    }

2.消费者

ps：初始化两个消费者消费test1和test2

package main
    
    import (
    	"time"
    	"fmt"
    	"log"
    	
    	"github.com/nsqio/go-nsq"
    )
    
    func main() {
    	err := initConsumer("test1", "test-channel1", "127.0.0.1:4161")
    	if err != nil {
    		log.Fatal("init Consumer error")
    	}
    	err = initConsumer("test2","test-channel2","127.0.0.1:4161")
    	if err != nil {
    		log.Fatal("init Consumer error")
    	}
    	select {
    
    	}
    }
    
    
    type nsqHandler struct {
    	nsqConsumer      *nsq.Consumer
    	messagesReceived int
    }
    
    //处理消息
    func (nh *nsqHandler)HandleMessage(msg *nsq.Message) error{
    	nh.messagesReceived++
    	fmt.Printf("receive ID:%s,addr:%s,message:%s",msg.ID, msg.NSQDAddress, string(msg.Body))
    	fmt.Println()
    	return nil
    }
    
    func initConsumer(topic, channel, addr string) error {
    	cfg := nsq.NewConfig()
    	cfg.LookupdPollInterval = 3*time.Second
    	c,err := nsq.NewConsumer(topic,channel,cfg)
    	if err != nil {
    		log.Println("init Consumer NewConsumer error:",err)
    		return err
    	}
    
    	handler := &nsqHandler{nsqConsumer:c}
    	c.AddHandler(handler)
    
    	err = c.ConnectToNSQLookupd(addr)
    	if err != nil {
    		log.Println("init Consumer ConnectToNSQLookupd error:",err)
    		return err
    	}
    	return nil
    }

结果：

生产者：

![70][]

消费者：

![70 1][]

也可以通过nsqadmin   [http://localhost:4171/  ][http_localhost_4171_] 查看nsq相关信息

[https_nsq.io_deployment_installing.html]: https://nsq.io/deployment/installing.html
[https_github.com_nsqio_nsq]: https://github.com/nsqio/nsq
[https_github.com_nsqio_go-nsq]: https://github.com/nsqio/go-nsq
[70]: /images/20220512/5d0c3de82adb4d7087d653310d7311e9.png
[70 1]: /images/20220512/6780a4270dbf4e36a402a09b6374e7db.png
[http_localhost_4171_]: http://localhost:4171/