nsq简介

nsq是go实现的高性能消息队列，部署相当简单。

一.搭建nsq集群

1.拉取docker镜像

docker pull nsqio/nsq  #拉取nsq镜像
docker images          #查看镜像

2.启动nsqlookup服务

 docker run -d --name nsqlookupd -p 4160:4160 -p 4161:4161 nsqio/nsq /nsqlookupd

3.启动两个nsqd节点服务

nsqd是nsq里面保存数据的服务

• 在服务器192.168.6.100上启动第一个nsqd节点

  docker run --net host -v /media/guo/wd5400tb2/nsq:/media/nsq --name nsqd00 nsqio/nsq /nsqd --tcp-address :4150 --http-address :4151 --broadcast-address=192.168.6.100  --lookupd-tcp-address=192.168.6.100:4160 --data-path /media/nsq

-v 指定挂载目录 -v /media/guo/st7200tb2/nsq:/media/nsq，把本机的/media/guo/st7200tb2/nsq挂载到/media/nsq目录下面

• 在服务器192.168.6.100上启动第二个nsqd节点

docker run --net host -v /media/guo/st7200tb2/nsq:/media/nsq --name nsqd01 nsqio/nsq /nsqd --tcp-address :4250 --http-address :4251 --broadcast-address=192.168.6.100  --lookupd-tcp-address=192.168.6.100:4160 --data-path /media/nsq

--data-path /media/nsq指定nsqd数据保存目录

4.启动admin节点

docker run -d --name nsqadmin -p 4171:4171 nsqio/nsq /nsqadmin --lookupd-http-address=192.168.6.100:4161

浏览器打开：http://192.168.6.100:4171/nodes ，此时的NSQd Nodes应该有两个，如果看到说明环境部署成功，恭喜了

二.curl 创建topic和channel

1.创建topic

curl -X POST http://192.168.6.100:4151/topic/create?topic=test-topic

2.创建channel

curl -X POST 'http://192.168.6.100:4151/channel/create?topic=test-topic&channel=test-channel'

三.测试

使用curl+nsq_to_file测试

• 向nsq写入消息

curl -d 'hello world 1' 'http://192.168.6.100:4151/pub?topic=test-topic'

• 使用nsq_to_file消费nsq里面的消息

nsq_to_file --topic=test-topic --output-dir=/tmp --lookupd-http-address=192.168.6.100:4161

使用golang测试

先启动服务端，用于消费消息，其中代码里指定nsqlookup服务的地址

server 端代码(消费消息)

package main

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

type myMessageHandler struct{}

// HandleMessage implements the Handler interface.
func (h *myMessageHandler) HandleMessage(m *nsq.Message) error {
    if len(m.Body) == 0 {
        // Returning nil will automatically send a FIN command to NSQ to mark the message as processed.
        fmt.Printf("body is empty\n")
        return nil
    }

    fmt.Printf("------------>:%s\n", m.Body)
    //err := processMessage(m.Body)

    // Returning a non-nil error will automatically send a REQ command to NSQ to re-queue the message.
    return nil
}

func main() {
    // Instantiate a consumer that will subscribe to the provided channel.
    config := nsq.NewConfig()
    config.MaxInFlight = 2
    consumer, err := nsq.NewConsumer("test-topic", "test-channel", config)
    if err != nil {
        log.Fatal(err)
    }

    // Set the Handler for messages received by this Consumer. Can be called multiple times.
    // See also AddConcurrentHandlers.
    consumer.AddHandler(&myMessageHandler{})

    // Use nsqlookupd to discover nsqd instances.
    // See also ConnectToNSQD, ConnectToNSQDs, ConnectToNSQLookupds.
    err = consumer.ConnectToNSQLookupd("192.168.6.100:4161")
    if err != nil {
        log.Fatal(err)
    }

    // Gracefully stop the consumer.

    for {
        time.Sleep(time.Second)
    }

    consumer.Stop()
}

client端(生产消息)

启动client用于向nsq写入消息。其中代码里面指定了一个nsqd服务的地址

package main

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

func main() {
    config := nsq.NewConfig()
    producer, err := nsq.NewProducer("192.168.6.100:4250", config)
    if err != nil {
        log.Fatal(err)
    }   

    messageBody := []byte("hello")
    topicName := "test-topic"

    // Synchronously publish a single message to the specified topic.
    // Messages can also be sent asynchronously and/or in batches.
    err = producer.Publish(topicName, messageBody)
    if err != nil {
        log.Fatal(err)
    }   

    // Gracefully stop the producer.
    producer.Stop()
}