介绍

nsqd

nsqd是一个守护进程，是接收、缓冲和传递消息给客户端的守护进程
https://nsq.io/components/nsqd.html#statsd

nsqlookupd

nsqlookupd是维护所有nsqd状态，提供服务发现的守护进程。它可以为消费者找到特定topic下的nsqd，提供了自动发现的服务。他不维持持久状态，也不需要和其他nsqlookupd实力协调以满足查询。因此根据你的系统要求尽可能多的部署nsqlookupd节点。他们消耗的资源很少，我们建议是为每个数据中心至少运行3个集群

nsqadmin

一个实时监控集群状态、执行各种管理任务的web管理平台，启动nsqadmin指定nsqlookupd地址

NSQL架构

image.png
1.生产者连接nsqd,通过nsqd将topic消息写入nsqd
2.消费者下从nsqookupd查询nsqd地址，再连接nsqd,从nsqd中读取消息（单机模式直连nsqd）
3.nsqdlookup发现和维护nsqd的状态，建议部署三台
4.nsqadmin连接nsqdlookup进行web端提供可视化

NSQ消息类型介绍

每个nsqd实例旨在一次处理多个数据流。这些数据流称为“topics”，一个topic具有1个或多个“channels”。每个channel都会收到topic所有消息的副本，实际上下游的服务是通过对应的channel来消费topic消息。
topic和channel不是预先配置的。topic在首次使用时创建，方法是将其发布到指定topic，或者订阅指定topic上的channel。channel是通过订阅指定的channel在第一次使用时创建的。
topic和channel都相互独立地缓冲数据，防止缓慢的消费者导致其他chennel的积压（同样适用于topic级别）。
channel可以并且通常会连接多个客户端。假设所有连接的客户端都处于准备接收消息的状态，则每条消息将被传递到随机客户端。例如：
NSQ - 图2
而言之，消息是从topic -> channel（每个channel接收该topic的所有消息的副本）多播的，但是从channel -> consumers均匀分布（每个消费者接收该channel的一部分消息）。

nsq消息发送流程

NSQ - 图3

nsq特性

  • 消息默认不持久化，可以配置成持久化模式。nsq采用的方式时内存+硬盘的模式，当内存到达一定程度时就会将数据持久化到硬盘。
    • 如果将—mem-queue-size设置为0，所有的消息将会存储到磁盘。
    • 服务器重启时也会将当时在内存中的消息持久化。
  • 每条消息至少传递一次。
  • 消息不保证有序。

    安装

    详细参数配置地址
    https://nsq.io/components/nsqd.html
    https://nsq.io/components/nsqlookupd.html
    https://nsq.io/components/nsqadmin.html
    构建分布式地址
    https://nsq.io/deployment/topology_patterns.html
    1.下载地址
    https://nsq.io/deployment/installing.html
    2.解压
    tar -zxvf nsq-1.2.1.linux-amd64.go1.16.6.tar.gz
    3.移动目录
    mv tar -zxvf nsq-1.2.1.linux-amd64.go1.16.6 /tools/nsq
    4.启动nsqlookupd,默认http4161端口,tcp4160
    /tools/nsq/bin/nsqlookupd &
    5.启动nsqd，默认http4151端口,tcp4150
    /tools/nsq/bin/nsqd -broadcast-address=192.168.72.10 —lookupd-tcp-address=127.0.0.1:4160 &
    -broadcast-address:表示nsqd广播地址，默认为主机名，如果消费者通过nsqlookupd查询nsqd地址再去连接会使用主机名：端口进行连接，因此需要在消费者端配置主机名解析，或者直接修改广播地址为真是IP，否者消费时会出现io/time异常
    修改主机名解析
    windows:C:\Windows\System32\drivers\etc\hosts
    linux:/etc/hosts
    6.启动nsqadmin，默认http4171端口
    /tools/nsq/bin/nsqadmin —lookupd-http-address=127.0.0.1:4161 &
    7.测试发布消息
    curl -d ‘hello world 1’ ‘http://127.0.0.1:4151/pub?topic=test‘
    8.测试接收消息
    /tools/nsq/bin/nsq_to_file —topic=test —output-dir=/tmp —lookupd-http-address=127.0.0.1:4161
    9.web端访问
    http://127.0.0.1:4171

Go操作NSQ

安装

go get -u github.com/nsqio/go-nsq


生产者示例

  1. // nsq_producer/main.go
  2. package main
  4. import (
  5.     "fmt"
  6.     "github.com/nsqio/go-nsq"
  7. )
  9. // NSQ Producer Demo
  11. var producer *nsq.Producer
  13. // 初始化生产者
  14. func initProducer(str string) (err error) {
  15.     config := nsq.NewConfig()
  16.     producer, err = nsq.NewProducer(str, config)
  17.     if err != nil {
  18.         fmt.Printf("create producer failed, err:%v\n", err)
  19.         return err
  20.     }
  21.     return nil
  22. }
  24. func main() {
  25.     //TCP端口
  26.     nsqAddress := "192.168.72.10:4150"
  27.     err := initProducer(nsqAddress)
  28.     if err != nil {
  29.         fmt.Printf("init producer failed, err:%v\n", err)
  30.         return
  31.     }
  32.     err = producer.Publish("topic_demo", []byte("测试"))
  33.     if err != nil {
  34.         fmt.Printf("publish msg to nsq failed, err:%v\n", err)
  35.     }
  36.     //以下为从标准数据读取数据测试
  37.     //reader := bufio.NewReader(os.Stdin) // 从标准输入读取
  38.     //for {
  39.     //    data, err := reader.ReadString('\n')
  40.     //    if err != nil {
  41.     //        fmt.Printf("read string from stdin failed, err:%v\n", err)
  42.     //        continue
  43.     //    }
  44.     //    data = strings.TrimSpace(data)
  45.     //    if strings.ToUpper(data) == "Q" { // 输入Q退出
  46.     //        break
  47.     //    }
  48.     //    // 向 'topic_demo' publish 数据
  49.     //    //重点：TODO
  50.     //    err = producer.Publish("topic_demo", []byte(data))
  51.     //    if err != nil {
  52.     //        fmt.Printf("publish msg to nsq failed, err:%v\n", err)
  53.     //        continue
  54.     //    }
  55.     //}
  56. }

消费者示例

  1. // nsq_consumer/main.go
  2. package main
  4. import (
  5.     "fmt"
  6.     "os"
  7.     "os/signal"
  8.     "syscall"
  9.     "time"
  11.     "github.com/nsqio/go-nsq"
  12. )
  14. // NSQ Consumer Demo
  16. // MyHandler 是一个消费者类型
  17. type MyHandler struct {
  18.     Title string
  19. }
  21. // HandleMessage 是需要实现的处理消息的方法
  22. func (m *MyHandler) HandleMessage(msg *nsq.Message) (err error) {
  23.     fmt.Printf("%s recv from %v, msg:%v\n", m.Title, msg.NSQDAddress, string(msg.Body))
  24.     return
  25. }
  27. // 初始化消费者
  28. func initConsumer(topic string, channel string, address string) (err error) {
  29.     config := nsq.NewConfig()
  30.     //配置每15秒查询一次nsqd可用列表
  31.     config.LookupdPollInterval = 15 * time.Second
  32.     c, err := nsq.NewConsumer(topic, channel, config)
  33.     if err != nil {
  34.         fmt.Printf("create consumer failed, err:%v\n", err)
  35.         return
  36.     }
  37.     consumer := &MyHandler{
  38.         Title: "test-consumer",
  39.     }
  40.     c.AddHandler(consumer)
  42.     // if err := c.ConnectToNSQD(address); err != nil { // 直接连NSQD
  43.     if err := c.ConnectToNSQLookupd(address); err != nil { // 通过lookupd查询
  44.         return err
  45.     }
  46.     return nil
  48. }
  50. func main() {
  51.     err := initConsumer("topic_demo", "first", "192.168.72.10:4161")
  52.     if err != nil {
  53.         fmt.Printf("init consumer failed, err:%v\n", err)
  54.         return
  55.     }
  56.     c := make(chan os.Signal)        // 定义一个信号的通道
  57.     signal.Notify(c, syscall.SIGINT) // 转发键盘中断信号到c
  58.     <-c                              // 阻塞
  59. }