高性能消息队列之nsq

Posted 2022-05-31 zlixing

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

 

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"