RocketMQ Client for Node.js

This official Node.js client is a lightweight wrapper around rocketmq-client-cpp, a finely tuned CPP client.

下面是中文文档

1. 因为是基于 rocketmq-client-cpp 做的node扩展，所以默认是下载 ali oss 对应 package.json 中 cppSDKVersion 版本的 .so .a .dylib 暂不支持 windows 如果需要，自行下载rocketmq-client-cpp 编译windows版本，
2. 下载地址 rocketmq-client-cpp release 下载好之后，放到 阿里云oss地址，或其他地址（默认会去阿里oss下载，但大多数情况会失败，所以我是上传到了自己的oss，如果需要，还是要自己上传到自己的oss ）：

let REGISTRY_MIRROR =
    process.env.NODE_ROCKETMQ_REGISTRY ||
    "https://opensource-rocketmq-client.oss-cn-hangzhou.aliyuncs.com";

// linux
    // 中间的 UBUNTU/14.04 要根据自身情况决定，如果是 centos 相同道理，如有不清楚，请查看源码中 download_lib.js 文件，
    ${REGISTRY_MIRROR}/cpp-client/linux/2.1.0/UBUNTU/14.04/librocketmq.a
// darwin
    ${REGISTRY_MIRROR}/cpp-client/mac/2.1.0/librocketmq.dylib

3. 本地需要安装构建相关东西 apt-get install python make build-essential g++ 可能有差异，需根据自身情况，自行解决

4. 安装过程如果有编译错误问题，可联系 [email protected] 如果有问题，欢迎各位大佬指正

5. 源码 example 目录有用例

Notice 1: This client is still in dev version. Use it cautiously in production.

Notice 2: This SDK is now only support macOS and Ubuntu 14.04. Ubuntu 16+ is not supported and CentOS is not tested yet.

Installation

$ apt-get install python make build-essential g++
$ npm install --save nodejs-rocketmq-client

Examples

You may view example/producer.js and example/push_consumer.js for quick start.

Usage

Require this package first.

const { Producer, PushConsumer } = require("nodejs-apache-rocketmq");

Producer

Constructor

new Producer(groupId[, instanceName][, options]);

Producer's constructor receives three parameters:

• groupId: the group id of the producer;
• instanceName: the instance name of the producer, optional;
• options: the options object, optional;
  • nameServer: the name server of RocketMQ;
  • groupName: the group name of this producer;
  • compressLevel: the compress level (0-9) of this producer, default to 5 where 0 is fastest and 9 is most compressed;
  • sendMessageTimeout: send message timeout millisecond, default to 3000 and suggestion is 2000 - 3000ms;
  • maxMessageSize: max message size with unit (B), default to 1024 * 128 which means 128K;
  • logFileNum: C++ core logic log file number, default to 3 and log file path is $HOME/logs/rocketmq-cpp;
  • logFileSize: size of each C++ core logic log file with unit (B);
  • logLevel: C++ core logic log level in "fatal", "error", "warn", "info", "debug", "trace" and "num".

e.g.

const { Producer } = require("nodejs-apache-rocketmq");
const producer = new Producer("GROUP_ID", "INSTANCE_NAME", {
    nameServer: "127.0.0.1:9876",
});

start

producer.start([callback]);

.start receives a callback function. If no callback passed, this function will return a Promise object.

e.g.

producer.start(function(err) {
    if(err) {
        //
    }
});

// or

producer.start().then(() => {
    //
}).catch(err => {
    //
});

shutdown

producer.shutdown([callback]);

.shutdown receives a callback function. If no callback passed, this function will return a Promise object.

e.g.

producer.shutdown(function(err) {
    if(err) {
        //
    }
});

// or

producer.shutdown().then(() => {
    //
}).catch(err => {
    //
});

send

producer.send(topic, body[, options][, callback]);

.send receives 4 parameters including a callback. If no callback passed, this function will return a Promise object.

• topic: the topic string;
• body: the message body string;
• options: the options object, optional;
  • keys: the keys for this message;
  • tags: the tags for this message;
• callback: the callback function, optional.

e.g.

producer.send("test", `baz ${i}`, {
    keys: "foo",
    tags: "bar"
}, function(err, result) {
    if(err) {
        // ...    
    } else {
        console.log(result);

        // console example:
        //
        //  { status: 0,
        //    statusStr: 'OK',
        //    msgId: '0101007F0000367E0000309DD68B0700',
        //    offset: 0 }
    }
});

send status and statusStr

| status | statusStr | | -------- | --------------------- | | 0 | OK | | 1 | FLUSH_DISK_TIMEOUT | | 2 | FLUSH_SLAVE_TIMEOUT | | 3 | SLAVE_NOT_AVAILABLE |

PushConsumer

Constructor

new PushConsumer(groupId[, instanceName][, options]);

PushConsumer's constructor receives three parameters:

• groupId: the group id of the push consumer;
• instanceName: the instance name of the push consumer, optional;
• options: the options object, optional;
  • nameServer: the name server of RocketMQ;
  • threadCount: the thread number of underlying C++ logic;
  • maxBatchSize: message max batch size;
  • logFileNum: C++ core logic log file number, default to 3 and log file path is $HOME/logs/rocketmq-cpp;
  • logFileSize: size of each C++ core logic log file with unit (B);
  • logLevel: C++ core logic log level in "fatal", "error", "warn", "info", "debug", "trace" and "num".

e.g.

const { PushConsumer } = require("nodejs-apache-rocketmq");
const consumer = new PushConsumer("GROUP_ID", "INSTANCE_NAME", {
    nameServer: "127.0.0.1:9876",
    threadCount: 3
});

start

consumer.start([callback]);

.start receives a callback function. If no callback passed, this function will return a Promise object.

e.g.

consumer.start(function(err) {
    if(err) {
        //
    }
});

// or

consumer.start().then(() => {
    //
}).catch(err => {
    //
});

shutdown

consumer.shutdown([callback]);

.shutdown receives a callback function. If no callback passed, this function will return a Promise object.

e.g.

consumer.shutdown(function(err) {
    if(err) {
        //
    }
});

// or

consumer.shutdown().then(() => {
    //
}).catch(err => {
    //
});

subscribe

Add a subscription relationship to consumer.

consumer.subscribe(topic[, expression]);

.subscribe receives two parameters which the second parameter is optional.

• topic: The topic to be subscribed;
• expression: The additional expression to be subscribed, optional. e.g. *.

On Message Event

If you want to receive messages from RocketMQ Server, you should add a listener for message event which receives 2 parameters.

function YOUR_LISTENER(msg, ack) {
    //
}

• msg: the message object to be consumed;
• ack: the Acknowledge object, which has a .done() function.

msg object looks like:

{ topic: 'test',
  tags: 'bar',
  keys: 'foo',
  body: 'baz 7',
  msgId: '0101007F0000367E0000339DD68B0800' }

You may call ack.done() to tell RocketMQ that you've finished your message successfully which is same as ack.done(true). And you may call ack.done(false) to tell it that you've failed.

e.g.

consumer.on("message", function(msg, ack) {
    console.log(msg);
    ack.done();
});

Apache RocketMQ Community

• RocketMQ Community Projects

Contact Us

• Mailing Lists: https://rocketmq.apache.org/about/contact/
• Home: https://rocketmq.apache.org
• Docs: https://rocketmq.apache.org/docs/quick-start/
• Issues: https://github.com/apache/rocketmq-client-nodejs/issues
• Ask: https://stackoverflow.com/questions/tagged/rocketmq
• Slack: https://rocketmq-community.slack.com/

How to Contribute

Contributions are warmly welcome! Be it trivial cleanup, major new feature or other suggestion. Read this how to contribute guide for more details.

License

Apache License, Version 2.0 Copyright (C) Apache Software Foundation