nsq 是基于 Go语言开发出来的消息队列中间件，今天在windows上来安装一下基础环境。

一、下载可执行文件

点击下载
下载完成之后解压：

二、执行 nsqlookup

nsqlookupd是管理拓扑信息的守护进程。客户端查询nsqlookupd以发现特定主题的nsqd生产者，nsqd节点广播主题和通道信息。

H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>nsqlookupd
[nsqlookupd] 2022/12/21 18:02:36.729343 INFO: nsqlookupd v1.2.1 (built w/go1.16.6)
[nsqlookupd] 2022/12/21 18:02:36.739842 INFO: HTTP: listening on [::]:4161
[nsqlookupd] 2022/12/21 18:02:36.739842 INFO: TCP: listening on [::]:4160

三、执行 nsqd

nsqd 是接收消息、排队并将消息传递给客户机的守护进程。

H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>nsqd --lookupd-tcp-address=127.0.0.1:4160
[nsqd] 2022/12/21 18:06:30.371132 INFO: nsqd v1.2.1 (built w/go1.16.6)
[nsqd] 2022/12/21 18:06:30.380208 INFO: ID: 94
[nsqd] 2022/12/21 18:06:30.381257 INFO: NSQ: persisting topic/channel metadata to nsqd.dat
[nsqd] 2022/12/21 18:06:30.508955 INFO: TCP: listening on [::]:4150
[nsqd] 2022/12/21 18:06:30.508955 INFO: LOOKUP(127.0.0.1:4160): adding peer
[nsqd] 2022/12/21 18:06:30.508955 INFO: HTTP: listening on [::]:4151
[nsqd] 2022/12/21 18:06:30.510027 INFO: LOOKUP connecting to 127.0.0.1:4160
[nsqd] 2022/12/21 18:06:30.512820 INFO: LOOKUPD(127.0.0.1:4160): peer info {TCPPort:4160 HTTPPort:4161 Version:1.2.1 BroadcastAddress:FENG}

四、执行 nsqadmin

nsqadmin是一个Web UI，用于实时查看聚合的集群统计信息并执行各种管理任务。

H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>nsqadmin --lookupd-http-address=127.0.0.1:4161
[nsqadmin] 2022/12/21 18:07:45.818217 INFO: nsqadmin v1.2.1 (built w/go1.16.6)
[nsqadmin] 2022/12/21 18:07:45.827762 INFO: HTTP: listening on [::]:4171

五、测试

1. 发布一个初始消息

H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>curl -d "hello world 1" "http://127.0.0.1:4151/pub?topic=test"
OK
H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>

• 可以看到返回OK，执行成功。

六、启动 nsq_to_file

nsq_to_file 使用指定的主题/通道并写入以换行符分隔的文件，可选地滚动和/或压缩该文件。

H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>nsq_to_file --topic=test --output-dir=./tmp --lookupd-http-address=127.0.0.1:4161
2022/12/21 18:13:14 INF    1 [test/nsq_to_file] querying nsqlookupd http://127.0.0.1:4161/lookup?topic=test
2022/12/21 18:13:14 INF    1 [test/nsq_to_file] (FENG:4150) connecting to nsqd
[nsq_to_file] 2022/12/21 18:13:33.800381 INFO: [test/nsq_to_file] opening tmp/test.FENG.2022-12-21_18.log
[nsq_to_file] 2022/12/21 18:13:33.800381 INFO: [test/nsq_to_file] syncing 1 records to disk
2022/12/21 18:14:22 INF    1 [test/nsq_to_file] querying nsqlookupd http://127.0.0.1:4161/lookup?topic=test
2022/12/21 18:15:22 INF    1 [test/nsq_to_file] querying nsqlookupd http://127.0.0.1:4161/lookup?topic=test
2022/12/21 18:16:22 INF    1 [test/nsq_to_file] querying nsqlookupd http://127.0.0.1:4161/lookup?topic=test
2022/12/21 18:17:22 INF    1 [test/nsq_to_file] querying nsqlookupd http://127.0.0.1:4161/lookup?topic=test

七、推送更多消息

H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>curl -d "hello world 1" "http://127.0.0.1:4151/pub?topic=test"
OK
H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>curl -d "hello world 1" "http://127.0.0.1:4151/pub?topic=test"
OK
H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>curl -d "hello world 1" "http://127.0.0.1:4151/pub?topic=test"
OK
H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>curl -d "hello world 1" "http://127.0.0.1:4151/pub?topic=test"
OK
H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>curl -d "hello world 1" "http://127.0.0.1:4151/pub?topic=test"
OK
H:\我的开发环境\消息队列\nsq-1.2.1.windows-amd64.go1.16.6\bin>

八、通过界面查看消息统计

浏览器打开：http://127.0.0.1:4171/

• 可以看到，刚刚发布了11条消息

九、Go 语言验证

1. 消费者（Consumer）程序段

package main

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

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.
        // In the case, a message with an empty body is simply ignored/discarded.
        return nil
    }

    // do whatever actual message processing is desired.
    //err := processMessage(m.Body)
    err := func(message []byte) error {
        log.Println("接收到消息： " + string(message))

        return nil
    }(m.Body)

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

func main() {
    // Instantiate a consumer that will subscribe to the provided channel.
    config := nsq.NewConfig()
    consumer, err := nsq.NewConsumer("topic", "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("localhost:4161")
    if err != nil {
        log.Fatal(err)
    }

    // wait for signal to exit
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    <-sigChan

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

2. 生产者（Producer）程序段

package main

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

func main() {

    config := nsq.NewConfig()
    producer, err := nsq.NewProducer("127.0.0.1:4150", config)
    if err != nil {
        log.Fatal(err)
    }
    messageBody := []byte("你好呀，七镜")
    topicName := "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 when appropriate (e.g. before shutting down the service)
    producer.Stop()
}

3. 生产者程序段执行结果

   
   

4. 消费者程序段执行结果

   
   

---

算是走了遍消息队列的入门程序， 后面还需要好好研究。