nsqjs客户端，未立即从go-nsq服务器端接收消息

Question

trying to learn nsq, and following the examples from here golang example and here nsqjs. I am sending messages in server side doing w/ a for loop and go routines

var wg sync.WaitGroup
for i := 0; i < 100; i++ {
    wg.Add(1)
    go func(x int) {
        defer wg.Done()

        chanName := fmt.Sprintf("import_progress_587e6442ff74889098498f6e")
        m := map[string]interface{}{
            "body": map[string]interface{}{
                "progress": x,
            },
        }
        msg, _ := json.Marshal(m)

        req := NSQPubReq{
            Topic: chanName,
            Body:  msg,
        }
        if err := producer.Publish(req.Topic, req.Body); err != nil {
        }
        utils.Info(fmt.Sprintf("sent msg=%v", string(msg)))

    }(i)
}

wg.Wait()

but the problem is, on the clientside.

// channel = 'import_progress_587e6442ff74889098498f6e'
let reader = new nsq.Reader(channel, channel, {
    //lookupdHTTPAddresses: '<<IP>>:4161',
    maxInFlight: 10000,
    snappy: true
})
reader.connect()

reader.on('message', (msg) => {
    var msgData = {
            id:     msg.id,
            body:   msg.body.toString(),
            chan:   channel
    }
    io.emit(channel, msgData)
    msg.finish()
})

the message don't come up immediately to the client. i will wait for a couple of seconds until the message come to the nodejs client. is there any settings that I need to do? thank you!

Answer 1

There are a couple of reasons why a nsqjs client will be slow to receive a message just published:

1. If the topic is new and the discovery of topics is via nsqlookupd, then by default, the nsqjs Reader will attempt to discover new topics every 30 seconds.

   From the example above, it looks like you are creating new topics for every import. I believe that if you start publishing messages first from the Golang client and then start the nsqjs client, then you shouldn't see the delay.

2. If you have multiple nsqds with a max-in-flight set too low, then it puts the nsqjs Reader in a starvation mode where it moves the RDY count between nsqds for a set period of time.

   I'm not sure that's what's going on here since I can't tell anything about the nsq topology. As long as your max-in-flight is set higher than the number of nsqd instances that you have, then you'll be in good shape.