高性能消息队列之nsq

青春须早为,岂能长少年。这篇文章主要讲述高性能消息队列之nsq相关的知识,希望能为你提供帮助。
NSQ 是实时的分布式消息处理平台,其设计的目的是用来大规模地处理每天数以十亿计级别的消息。
NSQ 具有分布式和去中心化拓扑结构,该结构具有无单点故障、故障容错、高可用性以及能够保证消息的可靠传递的特征。
NSQ 非常容易配置和部署,且具有最大的灵活性,支持众多消息协议。另外,官方还提供了拆箱即用 Go 和 python 库。如果读者有兴趣构建自己的客户端的话,还可以参考官方提供的协议规范。
网上有人翻译了国外的一篇文章:?  ??我们是如何使用NSQ处理7500亿消息的??
 
官网文档:  https://nsq.io/overview/quick_start.html">   https://nsq.io/overview/quick_start.html
中文文档:?  http://wiki.jikexueyuan.com/project/nsq-guide/"> ?  ?http://wiki.jikexueyuan.com/project/nsq-guide/?
 
【高性能消息队列之nsq】nsqd:基本的节点
nsqlookupd:汇总节点信息,提供查询和管理topic等服务

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

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

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

唯一性,在一个Nsq服务中只有一个nsqlookupd服务。当然也可以在集群中部署多个nsqlookupd,但它们之间是没有关联的
去中心化,即使nsqlookupd崩溃,也会不影响正在运行的nsqd服务
充当nsqd和naqadmin信息交互的中间件
提供一个http查询服务,给客户端定时更新nsqd的地址目录

 
nsqadmin:管理端展示UI界面,能有一个web页面去查看和操作
官方原话:是一套 WEB UI,用来汇集集群的实时统计,并执行不同的管理任务

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

提供一个对topic和channel统一管理的操作界面以及各种实时监控数据的展示,界面设计的很简洁,操作也很简单
展示所有message的数量,恩…装X利器
能够在后台创建topic和channel,这个应该不常用到
nsqadmin的所有功能都必须依赖于nsqlookupd,nsqadmin只是向nsqlookupd传递用户操作并展示来自nsqlookupd的数据

 

 
 
特性
默认一开始消息不是持久化的
nsq采用的方式时内存+硬盘的模式,当内存到达一定程度时就会将数据持久化到硬盘
如果将 ??--mem-queue-size?? 设置为 ??0??,所有的消息将会存储到磁盘。
是即使服务器重启也会将当时在内存中的消息持久化
消息是没有顺序的
这一点很关键,由于nsq使用内存+磁盘的模式,而且还有requeue的操作,所以发送消息的顺序和接收的顺序可能不一样
官方不推荐使用客户端发消息
官方提供相应的客户端发送消息,但是HTTP可能更方便一些
没有复制
nsq节点相对独立,节点与节点之间没有复制或者集群的关系。
没有鉴权相关模块
当前release版本的nsq没有鉴权模块,只有版本v0.2.29+高于这个的才有
几个小点
topic名称有长度限制,命名建议用下划线连接
消息体大小有限制
 
nsq优点& 缺点
优点:
部署极其方便,没有任何环境依赖,直接启动就行
轻量没有过多的配置参数,只需要简单的配置就可以直接使用
性能高
消息不存在丢失的情况
缺点:
消息无顺序
节点之间没有消息复制
 
 
 
[安装]
Docker:
添加docker-compose.yml文件
version: 3
services:
nsqlookupd:
image: nsqio/nsq
command: /nsqlookupd
restart: always
ports:
- 4160:4160
- 4161:4161
nsqd:
image: nsqio/nsq
command: /nsqd --lookupd-tcp-address=nsqlookupd:4160 --data-path=/data --broadcast-address=192.168.148.45
restart: always
depends_on:
- nsqlookupd
volumes:
- ./nsqd/data:/data
ports:
- 4150:4150
- 4151:4151
nsqadmin:
image: nsqio/nsq
command: /nsqadmin --lookupd-http-address=nsqlookupd:4161
restart: always
depends_on:
- nsqlookupd
ports:
- 4171:4171

参数说明:
> -lookupd-tcp-address 为上面nsqlookupd的IP和tcp的端口4160
> -lookupd-http-address 是http的端口也就是4161因为admin通过http请求来查询相关信息


客户端写入消息:
curl -d hello world 1 http://127.0.0.1:4151/pub?topic=hello word

队列消息写入文件:
./nsq_to_file --topic=test --output-dir=/tmp/log --lookupd-http-address=127.0.0.1:4161

 
 
 
使用:
golang:
go get -u github.com/nsqio/go-nsq

生产者:
package main

import (
"github.com/nsqio/go-nsq"
"log"
)

func main()
//初始化生产者
config := nsq.NewConfig()
producer, err := nsq.NewProducer("127.0.0.1:4150", config)
if err != nil
panic(err)

msgBody := []byte("hello")
topName := "topic_test"
err = producer.Publish(topName, msgBody)
if err != nil
log.Fatal(err)

producer.Stop()


 
 
 
消费者:
创建消费者的方式有两种,一种是通过http请求来发现nsqd生产者和配置的topic,另一种是直接使用tcp请求来连接本地实例 。
package main

import (
"fmt"
"github.com/nsqio/go-nsq"
"log"
"os"
"os/signal"
"syscall"
)

type msgHandler struct

func processMsg(b []byte) error
content := string(b)
fmt.Printf("Msg: %v \\n", content)
return nil


// 必须实现该接口
func (h *msgHandler) HandleMessage(m *nsq.Message) error
if len(m.Body) == 0
return nil


err := processMsg(m.Body)
m.Finish()
return err



func main()
config := nsq.NewConfig()
consumer, err := nsq.NewConsumer("topic_test", "channel", config)
if err != nil
panic(err)


consumer.AddHandler(& msgHandler)
//err = consumer.ConnectToNSQLookupd("127.0.0.1:4161")
lookupAddr := []string
"127.0.0.1:4161",

err = consumer.ConnectToNSQLookupds(lookupAddr)
if err != nil
log.Fatal(err)


sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
< - sigChan

consumer.Stop()

 
 
 
常见问题:
1.error connecting to nsqd - dial tcp: i/o timeout
消费者链接nsqlooup超时错误
在nsq中,如果消费者通过nsqlookup去连接nsqd从而获取消息中的数据,那个他的过程是如下的:
首先启动一个nsqlookup节点,用来维护,管理,发现nsqd节点
然后启动nsqd节点,nsqd节点启动的时候需要制定nsqlookup节点的地址,以及一个自身的广播地址参数
再然后生产者向指定的nsqd节点里面写入数据
之后消费者从nsqlookup中指定的topic里面找到对应的nsqd节点,并根据之前启动nsqd节点的广播地址去连接nsqd节点
最后,消费者连上nsqd节点以后就能够获取里面的数据了
再回到我们的错误
发生错误的原因是我启动nsqd节点的时候使用的指令如下:
nsqd --lookupd-tcp-address=127.0.0.1:4160

这里没有指定参数-broadcast-address参数,这个参数的作用就是将nsqd自身的地址发送给nsqlookup节点
所以我们重启nsqd节点就好了
nsqd --lookupd-tcp-address=127.0.0.1:4160 -broadcast-address=nsqd ip

 
 
 
 
 
 
 
 
其他:
集群搭建:
version: 3
services:
nsqlookupd:
restart: always
image: nsqio/nsq
command: /nsqlookupd --broadcast-address=192.168.2.3
ports:
- "4160:4160"
- "4161:4161"
nsqd:
restart: always
image: nsqio/nsq
command: /nsqd --lookupd-tcp-address=192.168.2.3:4160 --broadcast-address=192.168.2.3 --tcp-address=:4150 --http-address=:4151
depends_on:
- nsqlookupd
ports:
- "4150:4150"
- "4151:4151"
nsqd2:
restart: always
image: nsqio/nsq
command: /nsqd --lookupd-tcp-address=192.168.2.3:4160 --broadcast-address=192.168.2.3 --tcp-address=:4250 --http-address=:4251
depends_on:
- nsqlookupd
ports:
- "4250:4250"
- "4251:4251"
nsqadmin:
restart: always
image: nsqio/nsq
command: /nsqadmin --lookupd-http-address=192.168.2.3:4161
depends_on:
- nsqlookupd
ports:
- "4171:4171"

 



    推荐阅读