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 的更多内容，感兴趣的读者朋友们可以阅读官方文档。