NSQ 笔记

1

$ nsqlookupd

1

$ nsqd --lookupd-tcp-address=127.0.0.1:4160

1

$ nsqadmin --lookupd-http-address=127.0.0.1:4161

1

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

1

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

1
2

$ curl -d 'hello world 2' 'http://127.0.0.1:4151/pub?topic=test'
$ curl -d 'hello world 3' 'http://127.0.0.1:4151/pub?topic=test'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// NewConfig:
// This must be used to initialize Config structs. The only valid way to create a Config is via NewConfig. Values can be set directly, or through Config.Set()
// c.Set("tls_v1", true)
// c.Set("tls-insecure-skip-verify", true)
// c.Set("tls-min-version", "tls1.2")
// c.Set("local_addr", "1.2.3.4:27015")
// c.Set("dial_timeout", "5s")
config := nsq.NewConfig()

//  After Config is passed into NewProducer the values are no longer mutable (they are copied).
p, err := nsq.NewProducer("127.0.0.1:4150", config)
if err != nil {
    log.Fatal(err)
}

// if err := p.Publish("fuji", []byte("X-T3")); err != nil {
// 	log.Fatal("publish error: " + err.Error())
// }
for {
    // synchronously publishes a message body to the specified topic
    if err := p.Publish("test", []byte("test message")); err != nil {
        log.Fatal("publish error: " + err.Error())
    }
    time.Sleep(1 * time.Second)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

config := nsq.NewConfig()

// NewConsumer creates a new instance of Consumer for the specified topic/channel
// After Config is passed into NewConsumer the values are no longer mutable (they are copied).
consumer, _ := nsq.NewConsumer("fuji", "channel1", config)

// AddHandler sets the Handler for messages received by this Consumer. This can be called
// multiple times to add additional handlers. Handler will have a 1:1 ratio to message handling goroutines.
consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
    log.Printf("收到: %v", message)
    log.Printf("Body: %v", string(message.Body))
    if string(message.Body) == "X-T3" {
        consumer.Stop()
    }
    return nil
}))

// ConnectToNSQLookupd adds an nsqlookupd address to the list for this Consumer instance.
//
// If it is the first to be added, it initiates an HTTP request to discover nsqd
// producers for the configured topic.
//
// A goroutine is spawned to handle continual polling.

err := consumer.ConnectToNSQLookupd("127.0.0.1:4161")
if err != nil {
    log.Panic("连接失败")
}
// func (r *Consumer) ConnectToNSQLookupds(addresses []string) error {

// ConnectToNSQD takes a nsqd address to connect directly to.
// func (r *Consumer) ConnectToNSQD(addr string) error {

// read from this channel to block until consumer is cleanly stopped
<-consumer.StopChan

// Stop will initiate a graceful stop of the Consumer (permanent)
//
// NOTE: receive on StopChan to block until this process completes
// func (r *Consumer) Stop() {