nsq入门

• nsq组件
• 快速入门
• nsqd.dat
• go-nsq
• Topic、Channel、Message
  • Topic（主题）
  • Channel（通道）
  • Topic与Channel的关系
  • Message（消息）
  • 总结
• 参考

nsq组件

快速入门

在一个shell中，启动nsqlookupd：

nsqlookupd

在另一个shell中，启动nsqd，并注册到指定lookupd:

nsqd --lookupd-tcp-address=127.0.0.1:4160

注意：如果您的系统主机名未解析为127.0.0.1，则添加--broadcast-address=127.0.0.1

• nsqd：启动NSQ的守护进程。
• --lookupd-tcp-address=127.0.0.1:4160：指定 nsqd 连接到运行在本地（127.0.0.1）的 nsqlookupd 服务，该服务监听在4160端口上。

这样做的目的是让 nsqlookupd 知道有一个 nsqd 实例可用，并且客户端可以通过查询 nsqlookupd 来发现这个 nsqd 实例，从而进行消息的发送和接收。

通常，你会先启动 nsqlookupd 服务，然后启动多个 nsqd 实例，每个实例都指向同一个 nsqlookupd 服务，以便它们可以相互发现并协同工作。

在另一个shell中，启动nsqadmin：

nsqadmin --lookupd-http-address=127.0.0.1:4161

• nsqadmin：启动 NSQ 的管理界面。
• --lookupd-http-address=127.0.0.1:4161：指定 nsqadmin 连接到运行在本地（127.0.0.1）的 nsqlookupd 服务，该服务通过 HTTP 协议监听在 4161 端口上。

命令行参数 --lookupd-http-address 用于指定 nsqadmin 与 nsqlookupd 服务的通信地址和端口。这个参数允许 nsqadmin 知道如何找到 nsqlookupd 服务，从而能够获取整个集群的状态信息。

发布初始消息（也在集群中创建主题）使用curl命令向指定的nsqd实例发送消息：

curl -d 'hello world 1' 'http://127.0.0.1:4151/pub?topic=test'

• curl：调用curl命令行工具。
• -d 'hello world 1'：使用-d选项指定要发送的数据，这里是字符串hello world 1。
• 'http://127.0.0.1:4151/pub?topic=test'：指定要发送数据的URL。这里指定了nsqd实例的地址和端口（127.0.0.1:4151），以及HTTP的/pub端点，用于发布消息。?topic=test指定了消息要发送到的topic名称为test。

执行这条命令后，如果nsqd实例正常运行，并且监听在4151端口上，它会将消息hello world 1发布到名为test的topic中。

最后，在另一个shell中，启动nsq_to_file：

nsq_to_file --topic=test --output-dir=/tmp --lookupd-http-address=127.0.0.1:4161

这个NSQ命令是用来启动nsq_to_file工具的，它是一个用于将NSQ中指定topic的消息持久化到本地文件系统上的实用程序。下面是这个命令中各参数的详细解释：

• nsq_to_file: 这是NSQ工具集中的一员，专门用于消费NSQ消息并将它们写入文件。它适用于需要将消息记录下来或者进行离线分析的场景。

• --topic=test: 这个参数指定了你要消费的NSQ主题（Topic）名称。在这个例子中，主题名为test。NSQ允许你按照不同的主题来组织和发布消息，这样消费者可以根据兴趣订阅不同的主题。

• --output-dir=/tmp: 指定了消息将被写入的本地目录。这里设置的是/tmp目录，意味着所有从test主题接收的消息都会被存储到系统的临时目录下。你可以根据实际情况更改此路径。

• --lookupd-http-address=127.0.0.1:4161: 此参数指定了NSQ查找器（lookupd）的HTTP地址，它是用于发现NSQ集群中生产者和消费者的组件。在这个例子中，查找器运行在同一台机器上（localhost，即127.0.0.1），监听的HTTP端口是4161。NSQ客户端（如nsq_to_file）会通过这个地址找到负责特定主题的NSQ节点并开始消费消息。

综上所述，这个命令的作用是从名为test的NSQ主题中获取消息，并将这些消息保存到本地的/tmp目录下。它通过连接到位于本地且监听4161端口的NSQ查找器来发现和连接到正确的消息队列节点。这是在需要持久化消息或进行消息备份、分析等场景下非常有用的一个操作。

将更多消息发布到nsqd：

curl -d 'hello world 2' 'http://127.0.0.1:4151/pub?topic=test'
curl -d 'hello world 3' 'http://127.0.0.1:4151/pub?topic=test'

nsqd.dat

{
    "topics": [
        {
            "channels": [
                {
                    "name": "nsq_to_file",
                    "paused": false
                }
            ],
            "name": "test",
            "paused": false
        },
        {
            "channels": [
                {
                    "name": "ch1",
                    "paused": false
                },
                {
                    "name": "ch2",
                    "paused": false
                }
            ],
            "name": "testTopic",
            "paused": false
        }
    ],
    "version": "1.2.1"
}

go-nsq

github.com/nsqio/go-nsq

producer.go

package main

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

func main() {
	config := nsq.NewConfig()
	producer, err := nsq.NewProducer("127.0.0.1:4150", config)

	if err != nil {
		log.Panic(err)
	}

	for i := 0; i < 1000; i++ {
		msg := fmt.Sprintf("num-%d", i)
		log.Println("Pub:" + msg)
		err = producer.Publish("testTopic", []byte(msg))
		if err != nil {
			log.Panic(err)
		}
		time.Sleep(time.Second * 1)
	}

	producer.Stop()
}

consumer.go

package main

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

func main() {
	wg := &sync.WaitGroup{}
	wg.Add(1000)

	config := nsq.NewConfig()
	consumer, _ := nsq.NewConsumer("testTopic", "ch", config)
	consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		log.Printf("Got a message: %s", message.Body)
		wg.Done()
		return nil
	}))

	// 1. 直连 nsqd
	//err := consumer.ConnectToNSQD("127.0.0.1:4150")

	// 2.通过 nsqlookupd 服务发现
	err := consumer.ConnectToNSQLookupd("127.0.0.1:4161")
	if err != nil {
		log.Panic(err)
	}
	wg.Wait()

}

两个消费者消费同一个Channel时，根据分配算法Channel中的数据分配给不同的消费者

例如：Ch2中有10份数据，此时有两个消费者连接Ch2进行消费，Ch2中的数据均匀分配给两个消费者，最终每个消费者都获取到5份数据

Topic、Channel、Message

在NSQ中，Topic和Channel是两个核心概念，它们共同构成了消息传递的基本架构，支撑着消息的发布、订阅以及分发流程。下面是对这两个概念的详细解释：

Topic（主题）

• 概念：Topic是NSQ中消息的逻辑分类标识。每个发布到NSQ的消息都必须指定一个Topic。Topic可以看作是一个广播频道，比如新闻广播、音乐广播等，订阅者根据自己的兴趣订阅不同的Topic来获取相应类型的消息。

• 创建：Topic在消息第一次被发布到NSQ时自动创建，或者当消费者首次订阅一个尚未存在的Topic时也会触发创建。这意味着NSQ采取按需创建Topic的策略，不需要提前手动配置。

• 作用：Topic作为消息的分类容器，允许生产者向其发送消息，而消费者则通过订阅相关的Topic来接收消息。一个Topic可以被多个生产者写入消息，这些消息会被广播到所有订阅了该Topic的Channel。

• 特点：Topic是动态创建的，即当有消息首次被发布到一个不存在的Topic时，NSQ会自动创建这个Topic。此外，一个Topic可以被多个生产者写入，也能被多个Channel订阅。

Channel（通道）

• 概念：Channel是Topic下的一个子概念，代表了一类特定的消费者群体。一个Topic可以拥有零个或多个Channel。每个Channel都是Topic消息的一个独立队列，拥有自己的一套消息处理逻辑和消费者集合。简而言之，Channel实现了Topic消息的进一步细分和负载均衡。

• 创建：Channel通常在消费者首次订阅某个Topic时自动创建。这意味着Channel的生命周期与订阅者的活动紧密相关，但也可以通过API手动创建和管理。

• 负载均衡：由于每个Channel都是Topic消息的全量副本，拥有相同Topic的不同Channel会各自接收全部消息，这使得不同的消费者群体可以在各自的Channel中独立处理消息，实现负载均衡。

• 持久化与限流：Channel还可以配置消息的持久化策略（是否存储到磁盘）、消息处理的并发度以及消息的重试策略等，为不同类型的消费者需求提供了灵活性。

• 命名：Channel的命名通常反映消费者的业务逻辑或者处理特性，帮助区分不同的消息处理流程。

• 特点：Channel是消息的物理队列，具有消息缓存（内存和磁盘）以及消息处理策略（如重试、超时等）。Channel的存在使得即使没有消费者在线，消息也不会丢失，而是被暂存起来直到被消费。此外，Channel的创建也是动态的，随着消费者的订阅行为而产生。

Topic与Channel的关系

• 一对多：一个Topic可以关联多个Channel，每个Channel都是独立的，拥有自己的消息队列和消费者。

• 消息复制：当消息被发送到一个Topic时，该消息会被复制到该Topic下的所有Channel中。这意味着，无论你订阅了该Topic下的哪个Channel，都能接收到该Topic的所有消息。

• 解耦与扩展：这种设计允许高度的解耦和灵活的扩展

  • 生产者只需关注Topic，而不用关心具体的消费者或消息如何被处理
  • 消费者则通过订阅不同的Channel来实现业务逻辑的隔离和资源的优化利用。

通过Topic和Channel的组合，NSQ实现了消息的高效分发、灵活路由和负载均衡，适应了现代分布式系统对消息队列的多样化需求。

Message（消息）

• 概念：Message是NSQ传输的基本单位，包含了实际的数据内容和一些元数据（如消息ID、发送时间戳等）。
• 作用：消息是生产者和消费者之间交换信息的载体。生产者通过将数据封装成消息发送到指定的Topic，而消费者则从Channel中取出并处理这些消息。
• 特点：每个消息都有一个唯一的标识符（Message ID），可以用来追踪消息的处理状态。NSQ提供了消息确认机制，确保消息至少被处理一次或仅被处理一次，以此来保证消息的可靠性传递。

总结

NSQ通过Topic、Channel和Message这三个核心元素，构建了一个灵活、可扩展且高可用的消息队列系统。Topic负责消息的分类，Channel实现了消息的灵活路由和负载均衡，而Message则是实际传输的数据单元。这一架构设计支持了多种复杂的消息传递模式，满足了现代分布式系统中的多种消息处理需求。

参考

https://www.cnblogs.com/Alight/p/17272515.html