介绍
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
生产者
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 的更多内容,感兴趣的读者朋友们可以阅读官方文档。
推荐阅读:
参考资料:
https://nsq.io/overview/design.html
https://pkg.go.dev/github.com/nsqio/go-nsq@v1.0.8