go实现NSQ消息队列的集群部署

本文主要是介绍go实现NSQ消息队列的集群部署，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1 安装

官方下载页面根据自己的平台下载并解压即可。
我安装的是windows版本的
在这里插入图片描述
2 NSQ的工作模式
每个nsqd实例旨在一次处理多个数据流。这些数据流称为“topics”，一个topic具有1个或多个“channels”。每个channel都会收到topic所有消息的副本，实际上下游的服务是通过对应的channel来消费topic消息。

topic和channel不是预先配置的。topic在首次使用时创建，方法是将其发布到指定topic，或者订阅指定topic上的channel。channel是通过订阅指定的channel在第一次使用时创建的。

topic和channel都相互独立地缓冲数据，防止缓慢的消费者导致其他chennel的积压（同样适用于topic级别）。
在这里插入图片描述

3 集群部署

启动的exe文件均在下载目录bin下

启动nsqlookupd

nsqlookupd.exe

nsqlookupd监听的是4160 4161对应tcp http端口

启动3个nsqd
我这里启动了三个nsqd，记得在.bin/datas下建立三个node目录，

如果启动单个测试，参考博客

一个端口就对应一个nsqd
第一个nsqd就算监听4150 4151端口的

nsqd.exe --lookupd-tcp-address 127.0.0.1:4160 --broadcast-address 127.0.0.1 --data-path=./datas/node1 -http-address 127.0.0.1:4151 -tcp-address 127.0.0.1:4150
nsqd.exe --lookupd-tcp-address 127.0.0.1:4160 --broadcast-address 127.0.0.1 --data-path=./datas/node2 -http-address 127.0.0.1:4251 -tcp-address 127.0.0.1:4250
nsqd.exe --lookupd-tcp-address 127.0.0.1:4160 --broadcast-address 127.0.0.1 --data-path=./datas/node3 -http-address 127.0.0.1:4351 -tcp-address 127.0.0.1:4350

测试代码

总共有四个生产者

第1个生产者发给端口4150 topic == "topic_demo11"的nsqd
第2个生产者发给端口4150 topic == "topic_demo12"的nsqd
第3个生产者发给端口4250 topic == "topic_demo11"的nsqd
第4个生产者发给端口4350 topic == "topic_demo11"的nsqd

生产者代码

// nsq_producer/main.go
package mainimport ("fmt""sync""time""github.com/nsqio/go-nsq"
)// NSQ Producer Demovar producer *nsq.Producervar wg sync.WaitGroup
var mutex sync.Mutex
var k int64// 初始化生产者
func initProducer(str string) (err error) {config := nsq.NewConfig()producer, err = nsq.NewProducer(str, config)if err != nil {fmt.Printf("create producer failed, err:%v\n", err)return err}return nil
}//向指定ip port topic生产数据
func producerSendsData(ip string, port int, topic string) {//生产者往4150的nsqd上的dtopic_demo发信息nsqAddress := fmt.Sprintf("%s:%d", ip, port)//fmt.Println(nsqAddress)//一个端口就对应一个nsqderr := initProducer(nsqAddress)if err != nil {fmt.Printf("init producer failed, err:%v\n", err)return}for i := 0; i < 20; i++ {mutex.Lock()data := fmt.Sprintf("生产者发的数据：%d", k)// 向 'topic_demo' publish 数据err = producer.Publish(topic, []byte(data))if err != nil {fmt.Printf("publish msg to nsq failed, err:%v\n", err)continue}mutex.Unlock()k += 1//time.Sleep(time.Second)time.Sleep(time.Microsecond * 200)fmt.Println(data, nsqAddress)}defer wg.Done()
}
func main() {k = 0ip := "127.0.0.1"wg.Add(4)//一个端口就对应一个nsqd//第一个nsqd 端口4150 topic := "topic_demo"port1 := 4150topic11 := "topic_demo11"go producerSendsData(ip, port1, topic11)//第一个nsqd 端口4150 topic02 := "topic_demo02"topic12 := "topic_demo12"go producerSendsData(ip, port1, topic12)//第2个nsqd 端口4250 topic := "topic_demo11"//第2个nsqd我同样向topic_demo11中传输数据port2 := 4250topic21 := "topic_demo11" //"topic_demo21"go producerSendsData(ip, port2, topic21)//第3个nsqd 端口4350 topic02 := "topic_demo02"port3 := 4350topic31 := "topic_demo11"go producerSendsData(ip, port3, topic31)wg.Wait()//这部分代码是从标准输入读取了发送的/*reader := bufio.NewReader(os.Stdin) // 从标准输入读取for {data, err := reader.ReadString('\n')if err != nil {fmt.Printf("read string from stdin failed, err:%v\n", err)continue}data = strings.TrimSpace(data)if strings.ToUpper(data) == "Q" { // 输入Q退出break}// 向 'topic_demo' publish 数据err = producer.Publish("topic_demo", []byte(data))if err != nil {fmt.Printf("publish msg to nsq failed, err:%v\n", err)continue}}*/
}

总共有2个消费者

第一个消费者去收topic==topic_demo11的数据
第一个消费者去收topic==topic_demo12的数据

消费者代码

package mainimport ("fmt""os""os/signal""sync""syscall""time""github.com/nsqio/go-nsq"
)// NSQ Consumer Demo
var wg sync.WaitGroup// MyHandler 是一个消费者类型
type MyHandler struct {Title string
}// HandleMessage 是需要实现的处理消息的方法
func (m *MyHandler) HandleMessage(msg *nsq.Message) (err error) {fmt.Printf("%s recv from %v, msg:%v\n", m.Title, msg.NSQDAddress, string(msg.Body))return
}// 初始化消费者
func initConsumer(topic string, channel string, address string) {config := nsq.NewConfig()config.LookupdPollInterval = 15 * time.Secondc, err := nsq.NewConsumer(topic, channel, config)if err != nil {fmt.Printf("create consumer failed, err:%v\n", err)return}consumer := &MyHandler{Title: topic + " " + channel,}c.AddHandler(consumer)// if err := c.ConnectToNSQD(address); err != nil { // 直接连NSQDif err := c.ConnectToNSQLookupd(address); err != nil { // 通过lookupd查询fmt.Printf("c.ConnectToNSQLookupd(address) failed, err:%v\n", err)return}//time.Sleep(time.Second)defer wg.Done()}func main() {wg.Add(2)//topic_demo就是topic second是channel//第一个nsqd 端口4150 topic := "topic_demo11"下的数据用两个消费者（first和second）消费go initConsumer("topic_demo11", "first", "127.0.0.1:4161")go initConsumer("topic_demo12", "first", "127.0.0.1:4161")// //第一个nsqd 端口4150 topic02 := "topic_demo02"下的数据用1个消费者（first）消费// go initConsumer("topic_demo12", "first", "127.0.0.1:4161")// //第2个nsqd 端口4250 topic := "topic_demo21"// go initConsumer("topic_demo21", "second", "127.0.0.1:4161")wg.Wait()c := make(chan os.Signal)        // 定义一个信号的通道signal.Notify(c, syscall.SIGINT) // 转发键盘中断信号到c<-c                              // 阻塞
}