Docker
docker pull nsqio/nsq
nsqd
nsqd 是接收、队列和向客户端传递消息的守护进程。它可以独立运行，但通常在具有nsqlookupd实例的集群中进行配置（这种情况下，他将会发布主题和频道以便发现）
配置及api:https://nsq.io/components/nsqd.html
docker run --name nsqd -p 4150:4150 -p 4151:4151 nsqio/nsq /nsqd --broadcast-address=172.18.0.1
nsqlookup
nsqlookupd 是管理拓扑信息的守护进程。客户端可以查询 nsqlookupd 来发现特定主题的 nsqd 生产者，而 nsqd 节点会广播主题和通道的信息。
配置及api:https://nsq.io/components/nsqlookupd.html
nsqadmin
nsqadmin 是一个 Web 用户界面，可以实时查看汇总的集群统计信息，并执行各种管理任务。
docker run --name nsqadmin -p 4171:4171 nsqio/nsq /nsqadmin --nsqd-http-address=172.18.0.1:4151
Demo
// producer.go
package main
import (
"bufio"
"fmt"
"github.com/nsqio/go-nsq"
"os"
"strings"
)
var producer *nsq.Producer
func initProducer(str string) (err error) {
config := nsq.NewConfig()
producer, err = nsq.NewProducer(str, config)
if err != nil {
fmt.Println("create producer failed, err:", err)
return
}
return nil
}
func main() {
nsqAddress := "localhost:4150"
err := initProducer(nsqAddress)
if err != nil {
fmt.Printf("init producer failed, err:%v\n", err)
return
}
reader := bufio.NewReader(os.Stdin)
for {
data, err := reader.ReadString('\n')
if err != nil {
fmt.Printf("read string failed, err:%v\n", err)
continue
}
data = strings.TrimSpace(data)
if strings.ToUpper(data) == "Q" {
break
}
err = producer.Publish("topic_demo", []byte(data))
if err != nil {
fmt.Printf("publish msg to nsq failed, err:%v\n", err)
continue
}
}
}
// consumer.go
package main
import (
"fmt"
"time"
"github.com/nsqio/go-nsq"
)
type ConsumerT struct{}
func main() {
InitConsumer("topic_demo", "localhost:4150")
for {
time.Sleep(1 * time.Second)
}
}
func (*ConsumerT) HandleMessage(msg *nsq.Message) error {
fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
return nil
}
func InitConsumer(topic string, address string) {
cfg := nsq.NewConfig()
cfg.LookupdPollInterval = 15 * time.Second
c, err := nsq.NewConsumer(topic, "ch", cfg)
if err != nil {
panic(err)
}
c.SetLogger(nil, 0)
c.AddHandler(&ConsumerT{})
if err := c.ConnectToNSQD(address); err != nil {
panic(err)
}
}