Golang 语言编写的消息队列 NSQ 官方客户端 go-nsq 怎么使用？

01

介绍

NSQ 是 Golang 语言编写的实时分布式消息传递平台（也可以理解为消息队列），它主要由三个守护进程组成，分别是 nsqd、nsqlookupd 和 nsqadmin。其中 nsqd 是核心组成部分，它负责处理客户端的请求，比如生产、排序和消费消息等；nsqlookupd 负责管理集群拓扑信息和提供一个最终一致性的发现服务，nsqadmin 是一个 web 界面的管理平台，可以用于实时查看集群信息和执行其他管理操作。单个 nsqd可以含有很多 topic，每个 topic 可以含有很多 channel。

NSQ 支持跨平台和多语言客户端。使用 Mac 的读者朋友们可以使用 brew 方便的安装 NSQ。本文我们主要介绍 NSQ 官方提供的 golang 客户端 go-nsq。关于 NSQ 的更多内容，感兴趣的读者朋友们可以查阅官方文档，限于篇幅，本文不再赘述。

使用 go-nsq 操作 NSQ，需要安装 go-nsq 库，它提供了许多用于操作 NSQ 的函数和方法。

安装方式：

// Mac 安装 nsq
brew install nsq
// 安装 go-nsq
go get -u github.com/nsqio/go-nsq

02

生产者

go-nsq 包中的 Producer 类型，用于向 NSQ 发送消息。首先，需要调用函数 NewProducer 创建一个 Producer 实例，接收参数是 string 类型的 addr 和指针类型的 nsq.Config（NSQ 配置信息），返回结果是一个 Producer 实例的地址和 error。

需要注意的是，必须调用函数 NewConfig 返回一个指针类型的 nsq.Config，它包含默认配置信息。可以通过调用该实例的 Set 方法设置配置信息，并且必须在用于传参之前设置，否则设置的配置信息将不会生效。

Producer 包含很多方法，本文主要介绍四个方法，分别是 Ping、String、Publish 和 Stop。其中 Ping 方法用于检测 Producer 是否连接成功它配置的 nsqd；String 方法返回 Producer 连接的 nsqd 的地址；Publish 方法用于同步发送消息到指定的 topic；Stop 方法用于优雅地停止 Producer。

示例代码：

  // 默认配置信息
 config := nsq.NewConfig()

 // 创建生产者
 producer, err := nsq.NewProducer("127.0.0.1:4150", config)
 if err != nil {
  log.Fatal(err)
 }
 // 验证生成者连接是否成功
 err = producer.Ping()
 if err != nil {
  log.Fatal(err)
 }
 // 返回生产者地址
 producerAddr := producer.String()
 log.Printf("producerAddr:%v", producerAddr)
 messageBody := []byte("hello")
 topicName := "topic"
 // 同步发送消息到指定 topic
 err = producer.Publish(topicName, messageBody)
 if err != nil {
  log.Fatal(err)
 }
 producer.Stop()

运行结果：

2021/10/23 18:58:23 INF    1 (127.0.0.1:4150) connecting to nsqd
2021/10/23 18:58:23 producerAddr:127.0.0.1:4150
2021/10/23 18:58:23 INF    1 stopping
2021/10/23 18:58:23 INF    1 exiting router

03

消费者

go-nsq 包中的 Consumer 类型，用于从 NSQ 消费消息。首先，需要调用函数 NewConsumer 创建一个 Consumer 实例，接收参数是 string 类型的 topic 和 channel，指针类型的 nsq.Config（NSQ 配置信息），返回结果是一个 Consumer 实例的地址和 error。

需要注意的是，必须调用函数 NewConfig 返回一个指针类型的 nsq.Config，它包含默认配置信息。可以通过调用该实例的 Set 方法设置配置信息，并且必须在用于传参之前设置，否则设置的配置信息将不会生效。

Consumer 包含很多方法，本文主要介绍三个方法，分别是 Stats、AddHandler 和 ConnectToNSQD。其中 Stats 方法用于检索 Consumer 的当前连接和消息统计信息；AddHandler 用于为 Consumer 消费的消息设置处理函数，每个处理函数都独立运行在一个 goroutine 中，如果需要启动多个 goroutine 运行处理函数，可以多次调用 AddHandler；ConnectToNSQD 用于连接配置的 nsqd。

示例代码：

  config := nsq.NewConfig()
 // 创建 Consumer
 consumer, err := nsq.NewConsumer("topic", "channel", config)
 if err != nil {
  log.Fatal(err)
 }
 consumerStats := consumer.Stats()
 log.Printf("consumerStats:%+v", consumerStats)
 // 给 Consumer 添加处理器，可添加多个，每个 Handler 都运行在单独的 goroutine 中
 consumer.AddHandler(&myMessageHandler{})
 // 连接 nsqd
 err = consumer.ConnectToNSQD("127.0.0.1:4150")
 if err != nil {
  log.Fatal(err)
 }
 <-consumer.StopChan

运行结果：

2021/10/23 18:59:30 consumerStats:&{MessagesReceived:0 MessagesFinished:0 MessagesRequeued:0 Connections:0}
2021/10/23 18:59:30 INF    1 [topic/channel] (127.0.0.1:4150) connecting to nsqd
2021/10/23 18:59:30 hello

04

总结

本文主要介绍 Golang 语言编写的实时分布式消息平台 NSQ 的 golang 客户端 go-nsq。它是 NSQ 官方提供的 NSQ Golang 客户端。并且分别介绍了 Producer 和 Consumer 的简单使用方法，大家可以根据自己的业务需求，使用 go-nsq 灵活运用 NSQ。关于 go-nsq 的更多内容，感兴趣的读者朋友们可以阅读官方文档。

推荐阅读：

Golang 语言怎么编写测试代码？

Golang 语言中 kafka 客户端库 sarama

Golang 语言怎么使用 net/http 标准库开发 http 应用？

Golang 语言中怎么拦截系统信号和优雅退出 http server？

Golang 语言中 Context 的使用方式

参考资料： https://nsq.io/overview/design.html https://pkg.go.dev/github.com/nsqio/go-nsq@v1.0.8