如题 定时消息文档

Feature Request:

Support STS. We need to allow aliyun account A to access MQ instance in account B. After reading the docs, I found that MQ support RAM STS AssumeRole method to cross-account access.

Related doc: https://help.aliyun.com/document_detail/93844.html

有时候不想打印这些无效的信息

继 #60 问题回复之后，更新最新代码，使用公网区域下的实例（仅有 TCP 协议接入点），仍然出现错误。

$ node producer.js

{ MQClientException
    at MQClient._defaultHandler (/Users/bocai/study/npm/ali-ons/lib/mq_client_api.js:559:17)
    at MQClient.getTopicRouteInfoFromNameServer (/Users/bocai/study/npm/ali-ons/lib/mq_client_api.js:197:14)
    at <anonymous>
    at process._tickCallback (internal/process/next_tick.js:189:7)
    at Function.Module.runMain (module.js:696:11)
    at startup (bootstrap_node.js:204:16)
    at bootstrap_node.js:625:3 name: 'MQClientException', code: undefined }

@denghongcai @gxcsoccer 在阿里云控制台上新建了Topic的，前两天测试都是好的，能帮忙看下问题吗？

背景： 因为 Producer 并不会持续发送数据，所以希望在收到任务调度的时候，初始化 Producer 并处理异步任务进行发送，任务结束后，销毁 Producer 以免占用额外的资源（CPU，内存，socket 连接等）

问题： 使用 Producer 发送数据后，看到 socket 的建立以及周期任务的执行；但调用 .close() 方法后，日志显示已经 unregister producer 并且 producer close 了，但已建立的 socket 连接依然是 ESTAB 状态，且日志依然周期性的输出 mq 相关任务的执行

ali-one 版本： 3.6.5
node: 10.9.0

----------- 2019-06-19 22：23--------
排查发现是因为 _producerTable 里面除了自己注册上去的 producer 外，还有一个 defaultProducer。 在 close 的时候，自己的 producer 移除掉了，但是 defaultProducer 还在， 导致 mq_client 不能关闭 mq_client_api.

是否可以考虑在 close 时检查当前是否只剩一个 defaultProducer，如果是的话就将其一并关掉，如果还有其他 Producer 或 Consumer 则继续保留。

只拉到过一次后来就没有反应了都买的是阿里云ons

@gxcsoccer 为了控制消费速率，有个需求是主动拉取消息数量，这个现在有支持吗？

看example 只看到订阅消息，没看到如何消费

希望完善文档啊 操作说明 以及内部方法属性

PR 已提

sdk 版本: 2.0.3
node 版本: 8.4.0

需要根据某个属性做 sharding ，但是在 sdk 中似乎木有找到 sharding 功能相关的选项，不知道这个 feature 现在有支持不？

I got error log when connect to apache rq 4.4 . Dose ali-one support it ?

        com.aliyun.openservices.ons.api.exception.ONSClientException: Connect broker failed, Topic=DevStatus, msgId=null

See http://docs.aliyun.com/cn#/pub/ons/faq/exceptions&connect_broker_failed for further details.

at com.aliyun.openservices.ons.api.impl.rocketmq.ProducerImpl.checkProducerException(ProducerImpl.java:213)
at com.aliyun.openservices.ons.api.impl.rocketmq.ProducerImpl.send(ProducerImpl.java:136)
at com.aliyun.openservices.ons.api.bean.ProducerBean.send(ProducerBean.java:58)

需求

目前 async await 已经正式可以使用，所以支持 Promise 接口势在必行

版本：2.0.3

请问，在 consumer 订阅后，Ctrl + c 退出前，做了 consumer.close() 操作，但是消费的消息总数不正确了，可能是什么原因引起的呢？

并且还有一个现象是后续的消费情况不太均衡了。

当 broker对fetchConsumeOffset请求返回错误信息 “Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first”，rocket sdk并未做任何处理，仍是返回默认为-1的offset，直接影响新的消费组无法消费历史消息。

    long minOffset =
                    this.brokerController.getMessageStore().getMinOffsetInQuque(requestHeader.getTopic(),
                        requestHeader.getQueueId());
 
            if (minOffset <= 0
                    && !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(
                        requestHeader.getTopic(), requestHeader.getQueueId(), 0)) {
                responseHeader.setOffset(0L);
                response.setCode(ResponseCode.SUCCESS);
                response.setRemark(null);
            }
            else {
                response.setCode(ResponseCode.QUERY_NOT_FOUND);
                response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first");
            }

通过源码可以看到如果消费组信息并没有存储在broker上则默认返回该错误。因此对于大多数并未堆积过多数量消息的情况下，可以返回offset为0.

通过Consumer实例使用过滤tag无效：

consumer.subscribe('rocker_mq', 'rock_tig', async function(msg,mq,pg) {
    console.log(`开始消费消息, msgId: ${msg.msgId}, body: `);)
});

上述片段的回调无法执行。

consumer.subscribe('rocker_mq', '*', async function(msg,mq,pg) {
    if(msg.tags == 'rock_tig')
        console.log(`开始消费消息, msgId: ${msg.msgId}, body: `);)
});

只能采用在客户端进行过滤的策略，浪费带宽

在阿里云后台点击 消息查询->按 Topic 查询，选择某条历史消息，点击 消息详情，假如当前 topic 下有消费机在线，你能看到如下界面

此时点击消费验证，会出现如下界面，在这里我们可以选择特定的机器进行消息验证

但是系统会返回如下报错，事实上消费机是能够收到消息，并且正确消费的，如第一张图所示。
消息验证的功能，对于系统接入调试很有帮助，怀疑消息验证的的协议当前 SDK 还未实现。

在阿里云官方文档中，有提到：

网络访问说明

在使用消息队列 RocketMQ 时，请注意以下网络访问限制：

只有在同一个地域下的同一个实例中的 Topic 和 Group ID 才能互通，即某 Topic 是在哪个地域的哪个实例中创建的，它就只能被同样在该地域下的该实例中的创建的 Group ID 对应的生产端和消费端访问。

例如，当某 Topic 是创建在华北 2下的实例 A 中，那么该 Topic 只能被在华北 2下的实例 A 中创建的 Group ID 对应的生产端和消费端访问。

如果只是测试，或者需要在本地（非阿里云 ECS 服务器）使用消息队列 RocketMQ 的服务，请将 Topic 和 Group ID 都创建在“公网”地域下的实例中。 生产端和消费端可以部署在本地或者部署在任意地域的 ECS 上，前提是本地服务器或者相应的 ECS 需要能够访问公网。但注意遵循上一条原则，Topic 不能跨实例使用。

测试说明

在"公网"区域下创建了一个实例，该实例只有 TCP 协议接入点。

{ ResponseError: MQPushConsumer occurred an error socket hang up, GET http://MQ_INST_125036_BaXaRFt4.mq-internet-access.mq-internet.aliyuncs.com:80 -1 (connected: true, keepalive socket: false, socketHandledRequests: 1, socketHandledResponses: 0)
headers: {}
at createHangUpError (_http_client.js:342:15)
at Socket.socketOnEnd (_http_client.js:437:23)
at emitNone (events.js:111:20)
at Socket.emit (events.js:208:7)
at endReadableNT (_stream_readable.js:1064:12)
at _combinedTickCallback (internal/process/next_tick.js:139:11)
at process._tickCallback (internal/process/next_tick.js:181:9)
code: 'ECONNRESET',
name: 'ResponseError',
data: undefined,
path: '/',
status: -1,
headers: {},
res:
{ status: -1,
statusCode: -1,
statusMessage: null,
headers: {},
size: 0,
aborted: false,
rt: 101,
keepAliveSocket: false,
data: undefined,
requestUrls:
[ 'http://mq_inst_125036_baxarft4.mq-internet-access.mq-internet.aliyuncs.com:80/' ],
timing: null,
remoteAddress: '112.124.141.191',
remotePort: 80,
socketHandledRequests: 1,
socketHandledResponses: 0 } }

请问，这个是什么问题呢

阿里云消息队列RocketMQ接入点需要包涵reginid， ali-ons怎样指定？

跑本地的单元测试不通过, 而且是很简单的配置错误, 这样的代码质量令人堪忧

now, only retry message when consumer process failed. But in some scenarios, the application wants to retry the message directly.

[Wed Dec 04 2019 21:14:05 GMT+0800 (**标准时间)][pid: 7268][MQProducer] MQClientException: MQPushConsumer occurred an errorMQPushConsumer occurred an error [mq:client] updateAllTopicRouterInfo occurred error, Error Stack: MQClientException: MQPushConsumer occurred an error [mq:client] updateAllTopicRouterInfo occurred error, at MQClient._defaultHandler (/Users/huanhuiliu/Project/demo/node_modules/ali-ons/lib/mq_client_api.js:559:17) at MQClient.getTopicRouteInfoFromNameServer (/Users/huanhuiliu/Project/demo/node_modules/ali-ons/lib/mq_client_api.js:197:14) at processTicksAndRejections (internal/process/task_queues.js:93:5) Error Additions: code: undefined

我当前的consumer，单进程下，系统资源消耗 ～100M 内存，20% CPU，TPS 只有 30-40/s，并发性明显不够。

目前的重试好像是在consumer端的循环，并没有发送类似于RECONSUME_LATER的消息；另外，死信队列现在是不是也不支持呢？

变量多次实例化Producer会出现 以下问题：
No topic route info in name server for the topic: TBW102
See http://rocketmq.apache.org/docs/faq/ for further details.
See http://rocketmq.apache.org/docs/faq/ for further details.

如果是仅实例一次 client 就不会出现这个问题
如果producer 多次实例化就就会出现

class MQProducerService extends Service { async send(topic, tags, payload, opts) { const logger = this.ctx.logger; const producer = new Producer(opts); let sendResult; try { const msg = new Message(topic, tags, JSON.stringify(payload)); sendResult = await producer.send(msg); } catch (err) { logger.error(err); } return sendResult; } }

refer to:
https://help.aliyun.com/document_detail/43348.html

当一个consumer subscribe了一个主题之后，如何unsubscribe，以供后续使用？

嗨，你好。我用的是 "ali-ons": "^3.9.0",

• 我在本地日志记录该消息已经完成消费，但是在控制台查询消费轨迹却显示该消息尚未消费。
• 当我想对消息进行消费验证时也会验证失败

最近使用 ali-ons 提供的 http 方式接入, 吞吐量太差. 达不到预期.

期待官方继续开源 nodejs sdk

Message#delayTimeLevel? milliseconds?

对应code的文档可以在哪里看

Module not found: Error: Can't resolve 'child_process'

这是什么错 child_process 哪来的？

报错堆栈信息: ./node_modules/address/lib/address.js 5:12-36

你好，我在用公网的消息队列做测试，发消息的时候，一直在报公网的接入点无法正常访问。是我的使用方式有问题吗？翻遍了文档，也找不到具体问题在哪里。请求解答

现在可以收到消除处理消息，但是没有ack 方法去消耗消息，因此云平台一直是未消费状态

document

• https://help.aliyun.com/document_detail/29590.html?spm=a2c4g.11186623.6.633.27f445d9FJAedx

plz improve document.

当使用CONSUME_FROM_TIMESTAMP设置消费位点时，解析为时间戳的方法错误：

exports.parseDate = function(str) {
  return new Date(
    Number(str.slice(0, 4)),
    Number(str.slice(4, 6)) - 1,
    Number(str.slice(6, 8)),
    Number(str.slice(6, 8)),
    Number(str.slice(10, 12)),
    Number(str.slice(12)));
};

改为

exports.parseDate = function(str) {
  return new Date(
    Number(str.slice(0, 4)),
    Number(str.slice(4, 6)) - 1,
    Number(str.slice(6, 8)),
    Number(str.slice(8, 10)),
    Number(str.slice(10, 12)),
    Number(str.slice(12)));
};

nameserver会有多台备份，当有一台nameserver出现问题的时候，会获取另外一台进行处理

ali-ones没有这个容错，随机获取一台nameserver后，发现异常直接报错

https://github.com/ali-sdk/ali-ons/blob/master/lib/remoting_client.js#L169

consumerGroup 对应 consumer id吗
第一次使用不是很熟练 能否提供稍微详细的文档

单例测试，只包括begg和egg-metaq等最小基础环境，begg3.5.0可以订阅消息，3.5.1不能订阅消息

背景：2.0.4版本线上环境多个ons消费端服务集体OOMKill。
现象：多个消费者服务启动时间相同，但大约8-9小时后集体OOM，可确定无消息处理的消费者也会OOM。
测试：
1、ali-ons 2.0.4
2、只启动消费端服务，不生产消息进行测试且消费端服务只启动ons消费者subscribe，无任何其它逻辑。
3、服务刚启动时50M，17个小时后稳定在225-230M，高峰283M，测试环境容器限制最大300M，超过则重启服务；
目前看，在跑几个小时就该溢出了，所以怀疑这个版本仍然存在内存泄漏的问题或是ali-ons是否有最低内存要求？ 麻烦作者看看，谢谢。

refer to:

• https://help.aliyun.com/document_detail/49323.html
• https://help.aliyun.com/document_detail/43349.html

RT

RocketMq已经设置了autoCreateTopicEnable=true，即自动创建不存在的topic。
但是使用该SDK给一个不存在的topic发送消息时，还是会提示: [mq:client_api] get Topic [xxx] RouteInfoFromNameServer is not exist value。

消费失败后，如何从指定点位重新拉取数据，没有 ack 机制吗，只能重启服务，重新 consumer.subscribe 传入 consumeFromWhere 来拉取吗？

那假如一次性拉取 32 条，并行处理，发现第一条消费失败，那要再一次重新拉取 32 条，会有重复数据，设置成一次拉取 1 条的话，效率又不行。。。

使用过程中发送，在 Message含有 delay 参数的情况下，消费者和生产者各自获取到的 msgId 不一致！
测试代码如下：

'use strict';
const co = require('co');
const producer = require('./alionsProducer');

co(function *() {
    for (let i = 0; i < 15; i++) {
        // co(producer.send({ a: 1, b: 2 }, {
        //     delay: i,
        // }));
        let result = yield producer.send({a:i,b:2}, {
            delay: i
        });
        console.log(`第 ${i} 次：${JSON.stringify(result)}`);
    }
});

如果将 delay: i 换成 delay: 0 则接收到的 msgId 一致
send 是对 producer 封装的一个代码：

function *send(content, options) {
        let msg = new Message(
        options.topic || 'Message_test_11',
        options.tag || '*',
        JSON.stringify(content)
    );

    if(options.delay) {
        msg.delayTimeLevel = options.delay;
    }
   
    let result = yield producer.send(msg);
    return result;
}