1.下载nsq

Release 1.2.1 · nsqio/nsq · GitHub

 

 

Nsq服务端简介

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

nsqlookupd

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

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

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

• 唯一性，在一个Nsq服务中只有一个nsqlookupd服务。当然也可以在集群中部署多个nsqlookupd，但它们之间是没有关联的

• 去中心化，即使nsqlookupd崩溃，也会不影响正在运行的nsqd服务

• 充当nsqd和naqadmin信息交互的中间件

• 提供一个http查询服务，给客户端定时更新nsqd的地址目录 

nsqadmin

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

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

• 提供一个对topic和channel统一管理的操作界面以及各种实时监控数据的展示，界面设计的很简洁，操作也很简单

• 展示所有message的数量，恩....装X利器

• 能够在后台创建topic和channel，这个应该不常用到

• nsqadmin的所有功能都必须依赖于nsqlookupd，nsqadmin只是向nsqlookupd传递用户操作并展示来自nsqlookupd的数据

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

nsqd

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

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

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

• 对订阅了同一个topic，同一个channel的消费者使用负载均衡策略（不是轮询）

• 只要channel存在，即使没有该channel的消费者，也会将生产者的message缓存到队列中（注意消息的过期处理）

• 保证队列中的message至少会被消费一次，即使nsqd退出，也会将队列中的消息暂存磁盘上(结束进程等意外情况除外)

• 限定内存占用，能够配置nsqd中每个channel队列在内存中缓存的message数量，一旦超出，message将被缓存到磁盘中

• topic，channel一旦建立，将会一直存在，要及时在管理台或者用代码清除无效的topic和channel，避免资源的浪费

2.运行Nsq服务集群

首先启动 nsqlookupd.exe  

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

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

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

 

启动nqsadmin

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

 启动之后，通过 http://127.0.0.1:4171/ 可以访问这个管理页面， 默认使用4171端口

 

3.使用

这里有2种消费者的写法，第一种是直连nsqd（tcp长连接），第二种是通过nsqlookupd的http接口查询后长连接到nsqd, 显然第二种更易于分布式容错和高可用。这里我们都贴下代码

第一种nsqd直连

package mainimport ("flag""log""time""github.com/nsqio/go-nsq"
)func main() {go startConsumer()startProducer()
}var url stringfunc init() {//具体ip,端口根据实际情况传入或者修改默认配置flag.StringVar(&url, "url", "127.0.0.1:4150", "nsqd")flag.Parse()
}// 生产者
func startProducer() {cfg := nsq.NewConfig()producer, err := nsq.NewProducer(url, cfg)if err != nil {log.Fatal(err)}// 发布消息for {if err := producer.Publish("test", []byte("test message")); err != nil {log.Fatal("publish error: " + err.Error())}time.Sleep(1 * time.Second)}
}// 消费者
func startConsumer() {cfg := nsq.NewConfig()consumer, err := nsq.NewConsumer("test", "sensor01", cfg)if err != nil {log.Fatal(err)}// 设置消息处理函数consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {log.Println(string(message.Body))return nil}))// 连接到单例nsqdif err := consumer.ConnectToNSQD(url); err != nil {log.Fatal(err)}<-consumer.StopChan
}

第二种nsqlookupd方式

package mainimport ("flag""log""time""github.com/nsqio/go-nsq"
)func main() {go startConsumer()startProducer()
}var url string
var url1 stringfunc init() {//具体ip,端口根据实际情况传入或者修改默认配置flag.StringVar(&url, "url", "127.0.0.1:4150", "nsqd")         //tcpflag.StringVar(&url1, "url1", "127.0.0.1:4161", "nsqlookupd") //httpflag.Parse()
}// 生产者
func startProducer() {cfg := nsq.NewConfig()producer, err := nsq.NewProducer(url, cfg)if err != nil {log.Fatal(err)}// 发布消息for {if err := producer.Publish("test", []byte("test message")); err != nil {log.Fatal("publish error: " + err.Error())}time.Sleep(1 * time.Second)}
}// 消费者
func startConsumer() {cfg := nsq.NewConfig()consumer, err := nsq.NewConsumer("test", "sensor01", cfg)if err != nil {log.Fatal(err)}// 设置消息处理函数consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {log.Println(string(message.Body))return nil}))// nsqlookupd//[]stringif err := consumer.ConnectToNSQLookupds([]string{url1}); err != nil {log.Fatal(err)}<-consumer.StopChan
}