Typical strategies of using NSQ, in Node.js. It's a wrapper of official client library(nsqjs) with different strategies.
The nsqjs is handy, but it requires you to send message with a known nsqd address, which's neither impractical nor adhered to the principle of avoiding SPOF. The best practise is always using nsqlookup, but when you got a bunch of nsqds by lookup, which one should you choose to send the message? This module preprares some typical strategies for you.
This library has been upgraded to v2.x by a rewritting in TypeScript, for the users who is using the v1.x version, please refer to migraiton from v1.
npm install nsq-strategies
-
connectConfig: It can be specified with an array of nsqlookupd addresses or a single nsqd. -
option:strategy:PRODUCER_STRATEGY.ROUND_ROBIN|PRODUCER_STRATEGY.FAN_OUT(default:PRODUCER_STRATEGY.ROUND_ROBIN)- Other optional properties are exactly same with option in
Writerof nsqjs. Refer here for details.
produce(topic, msgs[, produceOptions])topic: NSQ Topicmsgs: NSQ messages, should use array in delay message.produceOptions:retrydelay: send delay message in given millisecond.strategy:PRODUCER_STRATEGY.ROUND_ROBIN|PPRODUCER_STRATEGY.FAN_OUT(default:PRODUCER_STRATEGY.ROUND_ROBIN)maxFanoutNodes: the maximum nodes with FAN_OUT strategy(ignored if the strategy is not FAN_OUT)
- The producer discovers the nsqd nodes from lookupd
- Every
producerpicks up a single nsqd in round-robin way and sends the message. - The producer doesn't take the topics into consideration for the round-robin pattern.
- Example
import { Producer } from 'nsq-strategies';
const p = new Producer({
lookupdHTTPAddresses: ['127.0.0.1:9011', '127.0.0.1:9012']
}, {
strategy: PRODUCER_STRATEGY.ROUND_ROBIN
});
await p.connect();
await p.produce('topic', 'message');
// with options
// enable retry with default node-promise-retry strategy,
// send NSQ message with 2-second delay
await p.produce('topic', 'message', { retry: true, delay: 2000 });
// with options
// enable retry with given node-promise-retry strategy,
// send NSQ message with 2-second delay
await p.produce('topic', 'message', { retry: {
retries: 3,
minTimeout: 300
}, delay: 2000 });- Diagram
*************************
********* *********
**** ****
*** +---------------+ +---------------+ ***
** | | | | **
* | nsqlookupd1 | | nsqlookupd2 | *
** | | | | **
*** +---------------+ +---------------+ ***
**** ****
********* *********
*************************
^
|
|
+------+------+
| |
| Producer |
| |
+-+----+----+-+
| | |
+--------------+ | +---------------+
| msg1 | msg3 |
| |msg2 |
v v v
+---+----+ +----+---+ +----+---+
| | | | | |
| nsqd1 | | nsqd2 | | nsqd2 |
| | | | | |
+--------+ +--------+ +--------+
- The producer discovers the nsqd nodes from lookupd
- Every
producerspreads the message to all nsqd nodes. - This stategy is mainly for delivery guarantee, it's not designed for pub-sub mode in nsq, note the message is duplicated among the nsqds, if you have a consumer(client) listening to it's topic, it will get the same message multiple times, if this is not expected you have to de-dupe in the consumer side or make the operation for the message idempotent.
- Example
import { Producer, PRODUCER_STRATEGY } from 'nsq-strategies';
const p = new Producer({
lookupdHTTPAddresses: ['127.0.0.1:9011', '127.0.0.1:9012']
}, {
strategy: PRODUCER_STRATEGY.FAN_OUT
});
await p.connect();
await p.produce('topic', 'message');- Diagram
*************************
********* *********
**** ****
*** +---------------+ +---------------+ ***
** | | | | **
* | nsqlookupd1 | | nsqlookupd2 | *
** | | | | **
*** +---------------+ +---------------+ ***
**** ****
********* *********
*************************
^
|
|
+------+------+
| |
| Producer |
| |
+-+----+----+-+
| | |
+--------------+ | +---------------+
| msg1 | msg1 |
| |msg1 |
v v v
+---+----+ +----+---+ +----+---+
| | | | | |
| nsqd1 | | nsqd2 | | nsqd2 |
| | | | | |
+--------+ +--------+ +--------+
This is useful for development or debugging.
import { Producer } from 'nsq-strategies';
const p = new Producer({
nsqdHost: '127.0.0.1',
tcpPort: 9031
});
await p.connect();
await p.produce(topic, 'message');Ordinarily you only need one producer in your application, you can use the singleton method for convenience.
import { Producer, PRODUCER_STRATEGY} from 'nsq-strategies';
const lookupdAddr = ['127.0.0.1:9011', '127.0.0.1:9012'];
const opt = { strategy: PRODUCER_STRATEGY.ROUND_ROBIN };
//singleton will call connect automatically
const producer = await Producer.singleton({ lookupdHTTPAddresses: lookupdAddr }, opt);
await p.produce(topic, 'some message');option:autoConnect: boolean (default: true)- Other optional properties can be refered to https://github.com/dudleycarr/nsqjs#new-readertopic-channel-options for parameters usages. Currently
Consumeris mostly a delegation ofReaderin nsqjs.
Example:
import { Consumer } from 'nsq-strategies';
const c = new Consumer('topic', 'channel', {
lookupdHTTPAddresses: ['127.0.0.1:9011', '127.0.0.1:9012']
});
// callback style
c.consume((msg) => {
console.log(msg.body.toString());
const result = handle(msg);
if (result) {
msg.finish();
} else {
msg.requeue(3000); //requeue with delay of 3000 milliseconds
}
});
// Observable style
const consumer$ = c.toRx();
consumer$.subscribe({
next: (msg) => handle(msg),
error: (err) => console.log(err),
complete: () => console.log('completed')
});There's an auto rennection mechanisim on prodcuer, which means if you lost the connection of the nsqd discovered, it will try to reconnect automatically in an exponential way until it's timeouted.
For every produce, you can set the exponential retry for the message.
const opt = {
retry: {
retries: 5,
factor: 3
//forever: true
}
};
await p.produce(topic, 'message', opt, );- retries: The maximum amount of times to retry, Default is 10.
- factor: The exponential factor to use. Default is 2.
- forever: Whether to retry forever, defaults to false.
- refer retry for more options.
Testing with message queue is not easy, of course you can test your codes with a booted nsq clusters with dockers like this, but you should focus on your business logic instead of the nsq itself. This library provides a fake simplistic in-memory queue with exact the same interfaces of the Producer/Consumer for your testing convenience.
import { Producer, Consumer, setMock } from 'nsq-strategies';
// CAVEAT: the setMock should be prior to the creation of Producer/Consumer
setMock(true);
const p = new Producer({
lookupdHTTPAddresses: ['127.0.0.1:9011', '127.0.0.1:9012']
});
await p.connect();
await p.produce('topic', 'message');
// The message sent by the producer above will flow to the same topic of consumers
const c = new Consumer('topic', 'channel', {
lookupdHTTPAddresses: ['127.0.0.1:9011', '127.0.0.1:9012']
});
const consumer$ = c.toRx();
consumer$.subscribe((msg) => console.log(msg.toString()));A few major changes from v1.x
- We moved to TypeScript.
- We abandoned the callback pattern and adopted aysnc/Consumer.
- The
Consumernow support Rxjs by returning an Observable. - The RestFul APIs for Nsqd/Lookup/LookupdCluster now return Axios Response.
Producer.ROUND_ROBIN/FAN_OUTare renamed toPRODUCER_STRATEGY.ROUND_ROBIN/FAN_OUT.
- Load balance strategy(pick the nsqd which has least topics)
- An option to config the retry times/backoff of nsqd reconnection
- Auto refresh the nsqd pool
MIT