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}

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

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>

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>

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()
}

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()
}