本文参考：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才能订阅消息。