使用NSQ(附Golang代码)

4,772 阅读5分钟

上篇文章已经了解了消息中间件相关的知识,这篇文章学习一下Golang语言编写的知名消息中间件NSQ

nsq最初是由bitly公司开源出来的一款简单易用的消息中间件,它可用于大规模系统中的实时消息服务,并且每天能够处理数亿级别的消息。它有以下特性:

  1. 分布式。它提供了分布式的、去中心化且没有单点故障的拓扑结构,稳定的消息传输发布保障,能够具有高容错和高可用特性。
  2. 易于扩展。它支持水平扩展,没有中心化的消息代理(Broker),内置的发现服务让集群中增加节点非常容易。
  3. 运维方便。它非常容易配置和部署,灵活性高。
  4. 高度集成。现在已经有官方的Golang、Python和JavaScript客户端,社区也有了其他各个语言的客户端库方便接入,自定义客户端也非常容易。

现在开始体验它~

安装

首先安装它,我在Mac上用Homebrew安装:

❯ brew install nsq

组件

nsq一共有四种组件

nsqlookupd

nsqlookupd是负责管理拓扑信息并提供最终一致性的发现服务的守护进程(daemon)。在终端1启动它:

❯ nsqlookupd
[nsqlookupd] 2019/07/18 11:42:16.876296 INFO: nsqlookupd v1.1.0 (built w/go1.11)
[nsqlookupd] 2019/07/18 11:42:16.876864 INFO: HTTP: listening on [::]:4161
[nsqlookupd] 2019/07/18 11:42:16.876868 INFO: TCP: listening on [::]:4160

默认HTTP接口监听4161,TCP接口监听4160。

nsqd

nsqd是一个负责接收、排队、投递消息给客户端的守护进程。客户端通过查询 nsqlookupd 来发现指定话题(topic)的nsqd生产者,nsqd节点会广播话题(topic)和通道(channel)信息。数据流模型如下:

单个nsqd可以有多个topic,每个topic可以有多个channel。channel接收这个topic所有消息的副本,从而实现多播分发,而channel上的每个消息被分发给它的订阅者,从而实现负载均衡。

在终端2启动nsqd:

❯ nsqd --lookupd-tcp-address=127.0.0.1:4160
...
[nsqd] 2019/07/18 11:47:46.427184 INFO: HTTP: listening on [::]:4151
[nsqd] 2019/07/18 11:47:46.427195 INFO: TCP: listening on [::]:4150
[nsqd] 2019/07/18 11:47:46.427203 INFO: LOOKUP(127.0.0.1:4160): adding peer
[nsqd] 2019/07/18 11:47:46.427355 INFO: LOOKUP connecting to 127.0.0.1:4160
...

nsqd通过tcp端口连接到了nsqlookupd,它自己在4151接受HTTP请求,在4150接受TCP请求。

nsqadmin

nsqadmin 是一套WEB管理UI,用来汇集集群的实时统计,并执行不同的管理任务。在终端3启动它:

❯ nsqadmin --lookupd-http-address=127.0.0.1:4161
[nsqadmin] 2019/07/18 11:54:23.125392 INFO: nsqadmin v1.1.0 (built w/go1.11)
[nsqadmin] 2019/07/18 11:54:23.128755 INFO: HTTP: listening on [::]:4171

浏览器打开http://localhost:4171就能访问了,需要注意,管理UI可以按需启动。

功能工具

安装nsq后会增加nsq_stat/nsq_tail/nsq_to_file等功能工具,这些实用程序以数据流的形式提供了通用功能和内部检查,稍后能体验到。

命令行体验

❯ curl -d 'hello world 1' 'http://127.0.0.1:4151/pub?topic=test'  # 在终端4执行
OK  # 发布消息到nsqd,用Rest API完成,看参数表示话题是test。由于还没有test这个话题,会先创建话题再接收消息。
❯ nsq_tail --lookupd-http-address=127.0.0.1:4161 --topic=test  # 在终端5执行,一会再来看
# 回到终端4
❯ curl -d 'hello world 2' 'http://127.0.0.1:4151/pub?topic=test'
OK
~
❯ curl -d 'hello world 3' 'http://127.0.0.1:4151/pub?topic=test'
OK
~
# 回到终端5
❯ nsq_tail --lookupd-http-address=127.0.0.1:4161 --topic=test
2019/07/18 12:06:24 Adding consumer for topic: test
...
hello world 2
hello world 3

回到终端5,可以看到接受(消费)了启动nsq_tail后发布的2个消息,这就是功能工具的作用,另外如nsq_to_file是把消息发到文件。

nsq的Go客户端使用

首先安装go-nsq:

❯ go get github.com/nsqio/go-nsq

先看生产者:

package main

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

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

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

	chars := []byte("ABCDEFGHIJKLMNOPQRSTUVWXYZ")

	for {
		buf := make([]byte, 4)
		for i := 0; i < 4; i++ {
			buf[i] = chars[rand.Intn(len(chars))]
		}
		log.Printf("Pub: %s", buf)
		err = w.Publish("test", buf)
		if err != nil {
			log.Panic(err)
		}
		time.Sleep(time.Second * 1)
	}

	w.Stop()
}

NewProducer的第一个参数就是nsqd的地址,在这里做了个无限for循环,每次随机4个byte发布到test话题里面。

接着看消费者代码:

package main

import (
	"log"
	"sync"

	"github.com/nsqio/go-nsq"
)

func main() {

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

	config := nsq.NewConfig()
	q, _ := nsq.NewConsumer("test", "ch", config)
	q.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		log.Printf("Got a message: %s", message.Body)
		wg.Done()
		return nil
	}))
	err := q.ConnectToNSQD("127.0.0.1:4150")
	if err != nil {
		log.Panic(err)
	}
	wg.Wait()

}

一开始通过sync.WaitGroup安排了1000个待执行的等待组,NewConsumer的第一个参数是话题test,第二是通道名字,然后用AddHandler添加一个消费处理函数,在处理函数中会打印这个消息。

现在就可以体验了,首先启动消费者,再启动发布者。

❯ go run consumer.go
2019/07/18 15:29:29 INF    1 [test/ch] (127.0.0.1:4150) connecting to nsqd
2019/07/18 15:29:37 Got a message: ZGBA
2019/07/18 15:29:38 Got a message: ICMR
2019/07/18 15:29:39 Got a message: AJWW
2019/07/18 15:29:40 Got a message: HTHC
2019/07/18 15:29:41 Got a message: TCUA
...
❯ go run producer.go
2019/07/18 15:29:36 INF    1 (127.0.0.1:4150) connecting to nsqd
2019/07/18 15:29:37 Pub: ZGBA
2019/07/18 15:29:38 Pub: ICMR
2019/07/18 15:29:39 Pub: AJWW
2019/07/18 15:29:40 Pub: HTHC
2019/07/18 15:29:41 Pub: TCUA
...

代码地址

原文地址: strconv.com/posts/use-n…

完整代码可以在这个地址找到。

延伸阅读

  1. nsq.io/overview/qu…