NSQ

2020-08-05 本文已影响0人强某某

本文参考：https://zhuanlan.zhihu.com/p/37081073

简介

NSQ是Go语言编写的，开源的内存分布式消息队列中间件
可以大规模地处理每天数以十亿计级别的消息
分布式和去中心化拓扑结构，无单点故障
Github地址：https://github.com/nsqio/nsq

应用场景

异步处理, 把非关键流程异步化，提高系统的响应时间和健壮性

1.jpg
应用解耦,通过消息队列

2.jpg
流量削峰

3.jpg

NSQ组件介绍

nsqd，负责消息接收、保存以及发送消息给消费者的进程
nsqlookupd，负责维护所有nsqd的状态，提供服务发现的进程
nsqadmin，是一个web管理平台，实时监控集群以及执行各种管理任务

NSQ架构介绍

4.jpg

Topic概念，对应一个具体的队列。比如订单数据放到order_queue这个topic
Channel概念，每个消费者对应一个channel。实现消息可重复消费。

5.jpg

topic消息的逻辑关键词

topic 是 NSQ 消息发布的逻辑关键词，可以理解为人为定义的一种消息类型。当程序初次发布带 topic 的消息时,如果 topic 不存在,则会在 nsqd中创建

producer消息的生产者/发布者

producer 通过 HTTP API 将消息发布到 nsqd 的指定 topic ，一般有 pub/mpub 两种方式， pub 发布一个消息， mpub 一个往返发布多个消息。
producer 也可以通过 nsqd客户端的 TCP接口将消息发布给 nsqd 的指定 topic 。
当生产者 producer 初次发布带 topic 的消息给 nsqd 时,如果 topic 不存在，则会在 nsqd 中创建 topic

channel消息传递的通道

当生产者每次发布消息的时候,消息会采用多播的方式被拷贝到各个 channel 中, channel 起到队列的作用。
channel 与 consumer(消费者) 相关，是消费者之间的负载均衡,消费者通过这个特殊的channel读取消息。
在 consumer 想单独获取某个 topic 的消息时，可以 subscribe(订阅)一个自己单独命名的 nsqd中还不存在的 channel, nsqd会为这个 consumer创建其命名的 channel
Channel 会将消息进行排列，如果没有 consumer读取消息，消息首先会在内存中排队，当量太大时就会被保存到磁盘中。可以在配置中配置具体参数。
一个 channel 一般会有多个 consumer 连接。假设所有已连接的 consumer 处于准备接收消息的状态，每个消息将被传递到一个随机的 consumer。
Go语言中的channel是表达队列的一种自然方式，因此一个NSQ的topic/channel，其核心就是一个存放消息指针的Go-channel缓冲区。缓冲区的大小由 --mem-queue-size 配置参数确定。

consumer消息的消费者

consumer 通过 TCPsubscribe 自己需要的 channel
topic 和 channel 都没有预先配置。 topic 由第一次发布消息到命名 topic 的 producer 创建或第一次通过 subscribe 订阅一个命名 topic 的 consumer 来创建。 channel 被 consumer 第一次 subscribe 订阅到指定的 channel 创建。
多个 consumersubscribe一个 channel，假设所有已连接的客户端处于准备接收消息的状态，每个消息将被传递到一个随机的 consumer。
NSQ 支持延时消息， consumer 在配置的延时时间后才能接受相关消息。
Channel在 consumer 退出后并不会删除，这点需要特别注意

概叙

NSQ推荐通过 nsqd 实例使用协同定位 producer，这意味着即使面对网络分区，消息也会被保存在本地，直到它们被一个 consumer读取。更重要的是， producer不必去发现其他的 nsqd节点，他们总是可以向本地 nsqd实例发布消息
一个 producer向它的本地nsqd发送消息，要做到这点，首先要先打开一个连接( NSQ 提供 HTTP API 和 TCP 客户端等2种方式连接到 nsqd)，然后发送一个包含 topic和消息主体的发布命令(pub/mpub/publish)，在这种情况下，我们将消息发布到topic上，消息会采用多播的方式被拷贝到各个 channel中, 然后通过多个 channel以分散到我们不同需求的 consumer中。
channel起到队列的作用。多个 producer产生的 topic消息在每一个连接 topic的 channel上进行排队。
每个 channel的消息都会进行排队，直到一个 consumer把他们消费，如果此队列超出了内存限制，消息将会被写入到磁盘中。 nsqd节点首先会向 nsqlookup 广播他们的位置信息，一旦它们注册成功， consumer将会从 nsqlookup 服务器节点上发现所有包含事件 topic的 nsqd节点。
每个 consumer向每个 nsqd主机进行订阅操作，用于表明 consumer已经准备好接受消息了。这里我们不需要一个完整的连通图，但我们必须要保证每个单独的 nsqd实例拥有足够的消费者去消费它们的消息，否则 channel会被队列堆着。

NSQ特性介绍

消息默认不持久化，可以配置成持久化
每条消息至少传递一次
消息不保证有序

NSQ搭建

6.jpg

解压之后如下

7.jpg

首先双击启动nsqlookupd.exe
.\nsqd.exe --lookupd-tcp-address=127.0.0.1:4160
.\nsqadmin.exe --lookupd-http-address=127.0.0.1:4161
nsqadmin其实就是UI界面，打开之后访问http://localhost:4171/即可
4160和4161其实是nsqlookupd.exe启动后打印的地址
生产者连接nqsd所以口链接端口是4150
消费者链接nsqlookupd，链接端口是4161

基本使用案例

消费者

package main

import (
    "fmt"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/nsqio/go-nsq"
)

// 消费者
type Consumer struct {
}

//处理消息
func (*Consumer) HandleMessage(msg *nsq.Message) error {
    fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
    return nil
}

// 主函数
func main() {
    err := initConsumer("order_queue", "first", "127.0.0.1:4161")
    if err != nil {
        fmt.Printf("init consumer failed, err:%v\n", err)
        return
    }
    c := make(chan os.Signal)
    signal.Notify(c, syscall.SIGINT)
    <-c
}

//初始化消费者
func initConsumer(topic string, channel string, address string) error {
    cfg := nsq.NewConfig()
    cfg.LookupdPollInterval = 15 * time.Second     //设置服务发现的轮询时间,例如新的nsq出现
    c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一个消费者
    if err != nil {
        return err
    }

    consumer := &Consumer{}
    c.AddHandler(consumer) // 添加消费者接口

    //建立NSQLookupd连接
    if err := c.ConnectToNSQLookupd(address); err != nil {
        return err
    }
    return nil
}

生产者

package main

import (
    "bufio"
    "fmt"
    "os"
    "strings"

    "github.com/nsqio/go-nsq"
)

var producer *nsq.Producer

//入口函数
func main() {
    //nsq的地址
    nsqAddress := "127.0.0.1:4150"
    err := initProducer(nsqAddress)
    if err != nil {
        fmt.Printf("init producer failed, err:%v\n", err)
        return
    }
    //读取控制台输入
    reader := bufio.NewReader(os.Stdin)
    for {
        data, err := reader.ReadString('\n')
        if err != nil {
            fmt.Printf("read string failed, err:%v\n", err)
            continue
        }

        data = strings.TrimSpace(data)
        if data == "stop" {
            break
        }

        err = producer.Publish("order_queue", []byte(data))
        if err != nil {
            fmt.Printf("publish message failed, err:%v\n", err)
            continue
        }
        fmt.Printf("publish data:%s succ\n", data)
    }
}

// 初始化生产者
func initProducer(str string) error {
    var err error
    config := nsq.NewConfig()
    producer, err = nsq.NewProducer(str, config)

    if err != nil {
        return err
    }
    return nil
}

总结：由上代码测试时候发现，必须先建立topic和channel，否则生产者发送的消息不会被topic接收，nqsadminui会提示必须创建channel才会显示具体信息。

而且生产者此时是向topic发送消息，而topic下面可以有多个channel，所以这个消息会一个channel一份。而消费者则指定了topic和channel才能订阅消息。

NSQ

简介