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 是否连接成功它配置的WordPress模板 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 的更多内容，感兴趣的读者朋友们可以阅读官方文档。

本文地址：http://www.bzve.cn/news/609a64898742.html