nsq介绍和使用

2020-04-11  本文已影响0人  哆啦在这A梦在哪

为什么要使用Nsq

最近一直在寻找一个高性能，高可用的消息队列做内部服务之间的通讯。一开始想到用zeromq，但在查找资料的过程中，意外的发现了Nsq这个由golang开发的消息队列，毕竟是golang原汁原味的东西，功能齐全，关键是性能还不错。其中支持动态拓展，消除单点故障等特性， 都可以很好的满足我的需求

下面上一张Nsq与其他mq的对比图，看上去的确强大。下面简单记录一下Nsq的使用方法

图片来自golang2017开发者大会

Nsq服务端

Nsq服务端简介

在使用Nsq服务之前，还是有必要了解一下Nsq的几个核心组件
整个Nsq服务包含三个主要部分

nsqlookupd

先看看官方的原话是怎么说：
nsqlookupd是守护进程负责管理拓扑信息。客户端通过查询 nsqlookupd 来发现指定话题（topic）的生产者，并且 nsqd 节点广播话题（topic）和通道（channel）信息

简单的说nsqlookupd就是中心管理服务，它使用tcp(默认端口4160)管理nsqd服务，使用http(默认端口4161)管理nsqadmin服务。同时为客户端提供查询功能

总的来说，nsqlookupd具有以下功能或特性

nsqadmin

官方原话：是一套 WEB UI，用来汇集集群的实时统计，并执行不同的管理任务

总的来说，nsqadmin具有以下功能或特性

nsqadmin默认的访问地址是http://127.0.0.1:4171/

nsqd

官方原话：nsqd 是一个守护进程，负责接收，排队，投递消息给客户端

简单的说，真正干活的就是这个服务，它主要负责message的收发，队列的维护。nsqd会默认监听一个tcp端口(4150)和一个http端口(4151)以及一个可选的https端口

总的来说，nsqd 具有以下功能或特性

这是官方的图，第一个channel(meteics)因为有多个消费者，所以触发了负载均衡机制。后面两个channel由于没有消费者，所有的message均会被缓存在相应的队列里，直到消费者出现

这里想到一个问题是，如果一个channel只有生产者不停的在投递message，会不会导致服务器资源被耗尽？也许nsqd内部做了相应处理，但还是要避免这种情况的出现

Nsq服务端与客户端的关系

了解nsqlookupd，nsqd与客户端中消费者和生产者的关系

消费者

消费者有两种方式与nsqd建立连接

生产者

生产者必须直连nsqd去投递message(网上说，可以连接到nsqlookupd，让nsqlookupd自动选择一个nsqd去完成投递，但是我用Producer的tcp是连不上nsqlookupd的，不知道http可不可以…)，

这里有一个问题就是如果生产者所连接的nsqd炸了，那么message就会投递失败，所以在客户端必须自己实现相应的备用方案

安装Nsq

方法一

go get github.com/tools/godep

执行完后检查godep是否已经安装在bin目录下，一般都会自动安装，如果没有，用go install手动安装下

go get github.com/bmizerany/assert

godep get github.com/bitly/nsq/...

如果安装成功，bin目录里就会出现一大堆nsq_…开头的可执行文件

方法二

运行Nsq

运行单机nsqd服务

nsqd是一个独立的服务，启动一个nsqd就可以完成message的收发，启动一个单机的nsqd，很简单

nsqd

客户端可以使用http，也可以使用tcp，这里我使用是官方的go-nsq包做客户端，使用tcp进行message的收发

运行Nsq服务集群

nsqlookupd


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


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

基于go-nsq的客户端实现

几个值得注意的地方

go_nsq-send.go

//Nsq发送测试

package main

import (

 "bufio"

 "fmt"

 "github.com/nsqio/go-nsq"

 "os"

)

var producer *nsq.Producer

// 主函数

func main() {

 strIP1 := "127.0.0.1:4150"

 strIP2 := "127.0.0.1:4152"

 InitProducer(strIP1)

 running := true

 //读取控制台输入

 reader := bufio.NewReader(os.Stdin)

 for running {

  data, _, _ := reader.ReadLine()

  command := string(data)

  if command == "stop" {

   running = false

  }

  for err := Publish("test", command); err != nil; err = Publish("test", command) {

   //切换IP重连

   strIP1, strIP2 = strIP2, strIP1

   InitProducer(strIP1)

  }

 }

 //关闭

 producer.Stop()

}

// 初始化生产者

func InitProducer(str string) {

 var err error

 fmt.Println("address: ", str)

 producer, err = nsq.NewProducer(str, nsq.NewConfig())

 if err != nil {

 panic(err)

 }

}

//发布消息

func Publish(topic string, message string) error {

 var err error

 if producer != nil {

  if message == "" { //不能发布空串，否则会导致error

   return nil

  }

  err = producer.Publish(topic, []byte(message)) // 发布消息

  return err

 }

 return fmt.Errorf("producer is nil", err)

}

go_nsq-receive.go

//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)

 // }

}
上一篇下一篇

猜你喜欢

热点阅读