diff --git a/.gitignore b/.gitignore index f3c24b5..9d110de 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,4 @@ node_modules example/.config.js .nyc_output/ .idea +.vscode diff --git a/README.md b/README.md index af11605..09e5b01 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ const consumer = new Consumer({ consumerGroup: 'your-consumer-group', // namespace: '', // aliyun namespace support // isBroadcast: true, + // consumeOrderly: true, }); consumer.subscribe(config.topic, '*', async msg => { @@ -76,6 +77,15 @@ const producer = new Producer({ // msg.setStartDeliverTime(Date.now() + 5000); const sendResult = await producer.send(msg); + // order message example: send msgs with the same orderId to same queue + const orderId = '******'; + // pass string to 2nd arg, same string will be send to same queue (aliyun-ons api) + const sendResult = await producer.send(msg, orderId); + // or you can write your own message queue seletor (RocketMQ api) + const sendResult = await producer.send(msg, messageQueues => { + let select = Math.max(Math.abs(hashCode(orderId)), 0); + return messageQueues[select % messageQueues.length]; + }); console.log(sendResult); })().catch(err => console.error(err)) ``` diff --git a/lib/consumer/mq_push_consumer.js b/lib/consumer/mq_push_consumer.js index 55b010f..2ebd178 100644 --- a/lib/consumer/mq_push_consumer.js +++ b/lib/consumer/mq_push_consumer.js @@ -40,6 +40,7 @@ const defaultOptions = { * 默认回溯到相对启动时间的半小时前 */ consumeTimestamp: utils.timeMillisToHumanString(Date.now() - 1000 * 60 * 30), + consumeOrderly: false, // 是否分区顺序消费 pullThresholdForQueue: 500, // 本地队列消息数超过此阀值，开始流控 pullInterval: 0, // 拉取消息的频率, 如果为了降低拉取速度，可以设置大于0的值 consumeMessageBatchMaxSize: 1, // 消费一批消息，最大数 @@ -47,7 +48,9 @@ const defaultOptions = { parallelConsumeLimit: 1, // 并发消费消息限制 postSubscriptionWhenPull: true, // 是否每次拉消息时，都上传订阅关系 allocateMessageQueueStrategy: new AllocateMessageQueueAveragely(), // 队列分配算法，应用可重写 - maxReconsumeTimes: 16, // 最大重试次数 + maxReconsumeTimes: 16, // 最大重试次数, + suspendCurrentQueueTimeMillis: 1000, // 顺序消费再次消费间隔 + lockInterval: 20000, // 顺序消费循环锁定间隔 }; class MQPushConsumer extends ClientConfig { @@ -104,6 +107,10 @@ class MQPushConsumer extends ClientConfig { return this.options.consumerGroup; } + get consumeOrderly() { + return this.options.consumeOrderly === true; + } + get messageModel() { return this.options.isBroadcast ? MessageModel.BROADCASTING : MessageModel.CLUSTERING; } @@ -128,8 +135,11 @@ class MQPushConsumer extends ClientConfig { this.logger.info('[mq:consumer] consumer started'); this._inited = true; - // 订阅重试 TOPIC + // 订阅重试 TOPIC & 顺序队列锁定 if (this.messageModel === MessageModel.CLUSTERING) { + if (this.consumeOrderly) { + this.startScheduledTask('lockAll', this.options.lockInterval, 1000); + } const retryTopic = MixAll.getRetryTopic(this.consumerGroup); this.subscribe(retryTopic, '*', async msg => { const originTopic = msg.retryTopic; @@ -153,6 +163,7 @@ class MQPushConsumer extends ClientConfig { */ async close() { this._isClosed = true; + await this.unlockAll(); await this.persistConsumerOffset(); this._pullFromWhichNodeTable.clear(); this._subscriptions.clear(); @@ -165,6 +176,45 @@ class MQPushConsumer extends ClientConfig { this.emit('close'); } + /** + * start a schedule task + * @param {String} name - method name + * @param {Number} interval - schedule interval + * @param {Number} [delay] - delay time interval + * @return {void} + */ + startScheduledTask(name, interval, delay) { + (async () => { + await this._sleep(delay || interval); + while (this._inited) { + try { + await this[name](); + } catch (err) { + this.logger.error(err); + } + await this._sleep(interval); + } + })(); + } + + // 按照brokerName归并MessageQueue集合 + buildProcessQueueTableByBrokerName() { + const brokerMqs = new Map(); + for (const key of this._processQueueTable.keys()) { + if (this._processQueueTable.get(key)) { + const messageQueue = this._processQueueTable.get(key).messageQueue; + const brokerName = messageQueue.brokerName; + let mqs = []; + if (brokerMqs.has(brokerName)) { + mqs = brokerMqs.get(brokerName); + } + mqs.push(messageQueue); + brokerMqs.set(brokerName, mqs); + } + } + return brokerMqs; + } + newOffsetStoreInstance() { if (this.messageModel === MessageModel.BROADCASTING) { if (this.options.persistent) { @@ -175,6 +225,161 @@ class MQPushConsumer extends ClientConfig { return new RemoteBrokerOffsetStore(this._mqClient, this.consumerGroup); } + async lock(messageQueue) { + const findBrokerResult = this._mqClient.findBrokerAddressInSubscribe(messageQueue.brokerName, MixAll.MASTER_ID, true); + if (!findBrokerResult) { + return; + } + try { + const mqs = [ messageQueue ]; + const lockedMq = await this._mqClient.lockBatchMQ(findBrokerResult.brokerAddr, this.consumerGroup, this._mqClient.clientId, mqs, 1000); + for (const messageQueue of lockedMq) { + const item = this._processQueueTable.get(messageQueue.key); + const processQueue = item && item.processQueue; + if (processQueue) { + processQueue.locked = true; + processQueue.lastLockTimestamp = Date.now(); + } + } + const lockOK = lockedMq.some(mq => mq.key === messageQueue.key); + this.logger.info('the message queue lock %s, %s %s', lockOK ? 'OK' : 'Failed', this.consumerGroup, messageQueue.key); + return lockOK; + } catch (e) { + this.logger.error('lockBatchMQ exception, %s', messageQueue.key); + } + } + + async unlock(messageQueue, delay) { + if (delay >= 0) { + await this._sleep(delay); + } + const findBrokerResult = this._mqClient.findBrokerAddressInSubscribe(messageQueue.brokerName, MixAll.MASTER_ID, true); + if (!findBrokerResult) { + return; + } + try { + const mqs = [ messageQueue ]; + await this._mqClient.unlockBatchMQ(findBrokerResult.brokerAddr, this.consumerGroup, this._mqClient.clientId, mqs, 1000); + this.logger.info('unlockBatchMQ ok, group %s, mq %s', this.consumerGroup, messageQueue.key); + } catch (e) { + this.logger.error('unlockBatchMQ exception, %s', messageQueue.key); + } + } + + /* eslint-disable no-loop-func */ + async lockAll() { + const brokerMqs = this.buildProcessQueueTableByBrokerName(); + const lockTasks = []; + for (const key of brokerMqs.keys()) { + const mqs = brokerMqs.get(key); + if (mqs && mqs.length >= 0) { + const brokerName = key; + const findBrokerResult = this._mqClient.findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, true); + if (!findBrokerResult) { + continue; + } + lockTasks.push( + (async () => { + try { + const lockOKMQSet = await this._mqClient.lockBatchMQ(findBrokerResult.brokerAddr, this.consumerGroup, this._mqClient.clientId, mqs, 1000); + lockOKMQSet.forEach(mq => { + const item = this._processQueueTable.get(mq.key); + const processQueue = item && item.processQueue; + if (processQueue != null) { + if (!processQueue.locked) { + this.logger.info('the message queue locked OK, Group: %s %s', this.consumerGroup, mq.key); + } + processQueue.locked = true; + processQueue.lastLockTimestamp = Date.now(); + } + }); + for (const messageQueue of mqs) { + if (!lockOKMQSet.some(lockOKMQ => lockOKMQ.key === messageQueue.key)) { + const item = this._processQueueTable.get(messageQueue.key); + const processQueue = item && item.processQueue; + if (processQueue) { + processQueue.locked = false; + this.logger.warn('the message queue locked Failed, Group: %s %s', this.consumerGroup, messageQueue.key); + } + } + } + } catch (e) { + this.logger.error('lockBatchMQ exception, %s', brokerName); + } + })() + ); + } + } + // wait until all of (success or fail) + await Promise.all(lockTasks); + return; + } + + async unlockAll() { + const brokerMqs = this.buildProcessQueueTableByBrokerName(); + const unlockTasks = []; + for (const key of brokerMqs.keys()) { + const mqs = brokerMqs.get(key); + if (mqs && mqs.length >= 0) { + const brokerName = key; + const findBrokerResult = this._mqClient.findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, true); + if (!findBrokerResult) { + continue; + } + unlockTasks.push( + this._mqClient.unlockBatchMQ(findBrokerResult.brokerAddr, this.consumerGroup, this._mqClient.clientId, mqs, 1000) + .catch(() => this.logger.error('unlockBatchMQ exception, %s', brokerName)) + ); + } + } + // wait until all of (success or fail) + await Promise.all(unlockTasks); + return; + } + + async tryLockLaterAndReconsume(messageQueue, processQueue, delayMills, args) { + await this._sleep(delayMills); + + const lockOK = await this.lock(messageQueue); + if (lockOK) { + this.submitConsumeRequestLater(messageQueue, processQueue, 10, args); + } else { + this.submitConsumeRequestLater(messageQueue, processQueue, 3000, args); + } + } + + async submitConsumeRequestLater(messageQueue, processQueue, suspendTimeMillis, args) { + let timeMillis = suspendTimeMillis; + if (timeMillis === -1) { + timeMillis = this.options.suspendCurrentQueueTimeMillis; + } + + if (timeMillis < 10) { + timeMillis = 10; + } else if (timeMillis > 30000) { + timeMillis = 30000; + } + + await this._sleep(timeMillis); + this.submitConsumeRequest(messageQueue, processQueue, false, args); + } + + // orderlyProcessChain里存有处理顺序消息的Promise链, 内存消耗最小且不会泄露(内部任务有超时机制) + submitConsumeRequest(messageQueue, processQueue, checkDispathToConsume, args) { + if (checkDispathToConsume && processQueue.consuming === true || !processQueue.msgCount) { + return; + } + const prevTask = Promise.resolve(processQueue.orderlyProcessChain); + const task = prevTask.then(async () => { + try { + await this._consumeMessageQueueOrderly(messageQueue, args); + } catch (e) { + this.logger.error('submitConsumeRequest task exception', e); + } + }); + processQueue.orderlyProcessChain = task; + } + /** * subscribe * @param {String} topic - topic @@ -196,6 +401,7 @@ class MQPushConsumer extends ClientConfig { const subscriptionData = this.buildSubscriptionData(this.consumerGroup, topic, subExpression); const tagsSet = subscriptionData.tagsSet; const needFilter = !!tagsSet.length; + const args = { needFilter, tagsSet, subExpression }; this.subscriptions.set(topic, { handler, subscriptionData, @@ -212,8 +418,22 @@ class MQPushConsumer extends ClientConfig { } // 消息消费循环 - while (!this._isClosed && this.subscriptions.has(topic)) { - await this._consumeMessageLoop(topic, needFilter, tagsSet, subExpression); + if (this.consumeOrderly) { + this.on('mq_changed', mqKey => { + if (this._isClosed) { + this.removeListener('mq_changed'); + } + const item = this._processQueueTable.get(mqKey); + const { messageQueue, processQueue } = item; + this.submitConsumeRequest(messageQueue, processQueue, true, args); + }); + } else { + while (!this._isClosed && this.subscriptions.has(topic)) { + const hasMsg = await this._consumeMessage(topic, needFilter, tagsSet, subExpression); + if (!hasMsg) { + await this.await(`topic_${topic}_changed`); + } + } } } catch (err) { @@ -222,7 +442,79 @@ class MQPushConsumer extends ClientConfig { })(); } - async _consumeMessageLoop(topic, needFilter, tagsSet, subExpression) { + async _consumeMessageQueueOrderly(mq, args) { + const { needFilter, tagsSet, subExpression } = args; + const item = this._processQueueTable.get(mq.key); + if (item) { + const pq = item.processQueue; + pq.consuming = true; + if (this.messageModel === MessageModel.BROADCASTING || (pq.locked && !pq.lockExpired)) { + let continueConsume = true; + const beginTime = Date.now(); + while (continueConsume) { + if (pq.droped) { + this.logger.warn("the message queue not be able to consume, because it's dropped. %s", mq.key); + break; + } + if (this.messageModel === MessageModel.CLUSTERING) { + if (!pq.locked) { + this.logger.warn('the message queue not locked, so consume later, %s', mq.key); + this.tryLockLaterAndReconsume(mq, pq, 10, args); + break; + } + if (pq.lockExpired) { + this.logger.warn('the message queue lock expired, so consume later, %s', mq.key); + this.tryLockLaterAndReconsume(mq, pq, 10, args); + break; + } + } + const interval = Date.now() - beginTime; + if (interval > 60000) { + this.submitConsumeRequestLater(mq, pq, 10, args); + break; + } + try { + let msgs; + if (this.parallelConsumeLimit > pq.msgCount) { + msgs = pq.msgList.slice(0, pq.msgCount); + } else { + msgs = pq.msgList.slice(0, this.parallelConsumeLimit); + } + for (const msg of msgs) { + const handler = this._subscriptions.get(msg.topic).handler; + if (!msg.tags || !needFilter || tagsSet.includes(msg.tags)) { + // 此处不可能失败 + await this.consumeSingleMsg(handler, msg, mq, pq, true); + } else { + this.logger.debug('[MQPushConsumer] message filter by tags=%s, msg.tags=%s', subExpression, msg.tags); + } + } + // 注意这里必须是批量确认 + const offset = pq.remove(msgs.length); + if (offset >= 0) { + this._offsetStore.updateOffset(mq, offset, true); + } + } catch (e) { + continue; + } + + if (!pq.msgCount) { + continueConsume = false; + } + } + } else { + if (pq.droped) { + this.logger.warn("the message queue not be able to consume, because it's dropped. %s", mq.key); + } else { + this.tryLockLaterAndReconsume(mq, pq, 100, args); + } + } + pq.consuming = false; + } + return; + } + + async _consumeMessage(topic, needFilter, tagsSet, subExpression) { const mqList = this._topicSubscribeInfoTable.get(topic); let hasMsg = false; if (mqList && mqList.length) { @@ -243,9 +535,9 @@ class MQPushConsumer extends ClientConfig { for (const msg of msgs) { const handler = this._subscriptions.get(msg.topic).handler; if (!msg.tags || !needFilter || tagsSet.includes(msg.tags)) { - consumeTasks.push(this.consumeSingleMsg(handler, msg, mq, pq)); + consumeTasks.push(this.consumeSingleMsg(handler, msg, mq, pq, false)); } else { - this.logger.debug('[MQPushConsumer] message filter by tags=, msg.tags=%s', subExpression, msg.tags); + this.logger.debug('[MQPushConsumer] message filter by tags=%s, msg.tags=%s', subExpression, msg.tags); } } // 必须全部成功 @@ -264,12 +556,10 @@ class MQPushConsumer extends ClientConfig { } } - if (!hasMsg) { - await this.await(`topic_${topic}_changed`); - } + return hasMsg; } - async consumeSingleMsg(handler, msg, mq, pq) { + async consumeSingleMsg(handler, msg, mq, pq, isOrder) { // 集群消费模式下，如果消费失败，反复重试 while (!this._isClosed) { if (msg.reconsumeTimes > this.options.maxReconsumeTimes) { @@ -286,24 +576,39 @@ class MQPushConsumer extends ClientConfig { } catch (err) { err.message = `process mq message failed, topic: ${msg.topic}, msgId: ${msg.msgId}, ${err.message}`; this.emit('error', err); - - if (this.messageModel === MessageModel.CLUSTERING) { - // 发送重试消息 - try { - // delayLevel 为 0 代表由服务端控制重试间隔 - await this.sendMessageBack(msg, 0, mq.brokerName, this.consumerGroup); - return; - } catch (err) { - this.emit('error', err); - this.logger.error('[MQPushConsumer] send reconsume message failed, fallback to local retry, msgId: %s', msg.msgId); - // 重试消息发送失败，本地重试 - await this._sleep(5000); + if (isOrder) { + if (msg.reconsumeTimes >= this.options.maxReconsumeTimes) { + try { + // 投入死信队列 + await this.sendMessageBackFailed(msg, this.consumerGroup, true); + return; + } catch (err) { + this.emit('error', err); + this.logger.error('[MQPushConsumer] send dead message failed, continue local retry, msgId: %s', msg.msgId); + // 投入死信队列失败 + } } - // 本地重试情况下需要给 reconsumeTimes +1 + await this._sleep(5000); msg.reconsumeTimes++; } else { - this.logger.warn('[MQPushConsumer] BROADCASTING consume message failed, drop it, msgId: %s', msg.msgId); - return; + if (this.messageModel === MessageModel.CLUSTERING) { + // 发送重试消息 + try { + // delayLevel 为 0 代表由服务端控制重试间隔 + await this.sendMessageBack(msg, 0, mq.brokerName, this.consumerGroup); + return; + } catch (err) { + this.emit('error', err); + this.logger.error('[MQPushConsumer] send reconsume message failed, fallback to local retry, msgId: %s', msg.msgId); + // 重试消息发送失败，本地重试 + await this._sleep(5000); + } + // 本地重试情况下需要给 reconsumeTimes +1 + msg.reconsumeTimes++; + } else { + this.logger.warn('[MQPushConsumer] BROADCASTING consume message failed, drop it, msgId: %s', msg.msgId); + return; + } } } } @@ -398,6 +703,27 @@ class MQPushConsumer extends ClientConfig { return; } + if (this.consumeOrderly) { + if (processQueue.locked) { + if (pullRequest.lockedFirst !== true) { + const offset = await this.computePullFromWhere(pullRequest.messageQueue); + const brokerBusy = offset < pullRequest.nextOffset; + this.logger.info('the first time to pull message, so fix offset from broker. pullRequest: %j NewOffset: %s brokerBusy: %s', + pullRequest, offset, brokerBusy); + if (brokerBusy) { + this.logger.info('the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: %j NewOffset: %s', + pullRequest, offset); + } + pullRequest.lockedFirst = true; + pullRequest.nextOffset = offset; + } + } else { + await this._sleep(this.options.pullTimeDelayMillsWhenException); + this.logger.info('pull message later because not locked in broker, %s', pullRequest.messageQueue.key); + return; + } + } + processQueue.lastPullTimestamp = Date.now(); const data = this.subscriptions.get(messageQueue.topic); const subscriptionData = data && data.subscriptionData; @@ -440,6 +766,7 @@ class MQPushConsumer extends ClientConfig { // submit to consumer processQueue.putMessage(pullResult.msgFoundList); this.emit(`topic_${messageQueue.topic}_changed`); + this.emit('mq_changed', messageQueue.key); break; } case PullStatus.NO_NEW_MSG: @@ -548,12 +875,13 @@ class MQPushConsumer extends ClientConfig { * @return {void} */ async doRebalance() { + const isOrder = this.consumeOrderly; for (const topic of this.subscriptions.keys()) { - await this.rebalanceByTopic(topic); + await this.rebalanceByTopic(topic, isOrder); } } - async rebalanceByTopic(topic) { + async rebalanceByTopic(topic, isOrder) { this.logger.info('[mq:consumer] rebalanceByTopic: %s, messageModel: %s', topic, this.messageModel); const mqSet = this._topicSubscribeInfoTable.get(topic); // messageQueue list if (!mqSet || !mqSet.length) { @@ -564,7 +892,7 @@ class MQPushConsumer extends ClientConfig { let changed; let allocateResult = mqSet; if (this.options.isBroadcast) { - changed = await this.updateProcessQueueTableInRebalance(topic, mqSet); + changed = await this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder); } else { const cidAll = await this._mqClient.findConsumerIdList(topic, this.consumerGroup); this.logger.info('[mq:consumer] rebalance topic: %s, with consumer ids: %j', topic, cidAll); @@ -575,7 +903,7 @@ class MQPushConsumer extends ClientConfig { allocateResult = this.allocateMessageQueueStrategy.allocate(this.consumerGroup, this._mqClient.clientId, mqSet, cidAll); this.logger.info('[mq:consumer] allocate queue for group: %s, clientId: %s, result: %j', this.consumerGroup, this._mqClient.clientId, allocateResult); - changed = await this.updateProcessQueueTableInRebalance(topic, allocateResult); + changed = await this.updateProcessQueueTableInRebalance(topic, allocateResult, isOrder); } } if (changed) { @@ -588,9 +916,10 @@ class MQPushConsumer extends ClientConfig { * update process queue * @param {String} topic - topic * @param {Array} mqSet - message queue set + * @param {Boolean} isOrder - is order or not * @return {void} */ - async updateProcessQueueTableInRebalance(topic, mqSet) { + async updateProcessQueueTableInRebalance(topic, mqSet, isOrder) { let changed = false; // delete unnecessary queue for (const key of this._processQueueTable.keys()) { @@ -619,6 +948,11 @@ class MQPushConsumer extends ClientConfig { continue; } + if (isOrder && !(await this.lock(messageQueue))) { + this.logger.warn('doRebalance, %s, add a new mq failed, %s, because lock failed', this.consumerGroup, messageQueue.key); + continue; + } + const nextOffset = await this.computePullFromWhere(messageQueue); // double check if (this._processQueueTable.has(messageQueue.key)) { @@ -627,6 +961,11 @@ class MQPushConsumer extends ClientConfig { if (nextOffset >= 0) { const processQueue = new ProcessQueue(); + if (isOrder) { + // this pq has been locked + processQueue.locked = true; + processQueue.lastLockTimestamp = Date.now(); + } changed = true; this._processQueueTable.set(messageQueue.key, { messageQueue, @@ -731,10 +1070,16 @@ class MQPushConsumer extends ClientConfig { async removeUnnecessaryMessageQueue(messageQueue) { await this._offsetStore.persist(messageQueue); this._offsetStore.removeOffset(messageQueue); - // todo: consume later ？ + if (this.consumeOrderly && this.messageModel === MessageModel.CLUSTERING) { + try { + // 异步unlock + this.unlock(messageQueue, 20000); + } catch (e) { + this.logger.error('removeUnnecessaryMessageQueue Exception'); + } + } } - async sendMessageBack(msg, delayLevel, brokerName, consumerGroup) { const brokerAddr = brokerName ? this._mqClient.findBrokerAddressInPublish(brokerName) : msg.storeHost; @@ -770,6 +1115,26 @@ class MQPushConsumer extends ClientConfig { } } + // 投入死信队列 + async sendMessageBackFailed(msg, consumerGroup, orderly) { + let newMsg; + const thatConsumerGroup = consumerGroup ? consumerGroup : this.consumerGroup; + if (!orderly && MixAll.isRetryTopic(msg.topic)) { + newMsg = msg; + } else { + newMsg = new Message(MixAll.getRetryTopic(thatConsumerGroup), '', msg.body); + newMsg.flag = msg.flag; + newMsg.properties = msg.properties; + newMsg.originMessageId = msg.originMessageId || msg.msgId; + newMsg.retryTopic = msg.topic; + newMsg.properties[MessageConst.PROPERTY_MAX_RECONSUME_TIMES] = this.options.maxReconsumeTimes; + } + + newMsg.properties[MessageConst.PROPERTY_RECONSUME_TIME] = String(orderly ? msg.reconsumeTimes : msg.reconsumeTimes + 1); + newMsg.delayTimeLevel = 3 + msg.reconsumeTimes; + await (await MQProducer.getDefaultProducer()).send(newMsg); + } + // * viewMessage(msgId) { // const info = MessageDecoder.decodeMessageId(msgId); // return yield this._mqClient.viewMessage(info.address, Number(info.offset.toString()), 3000); diff --git a/lib/mq_client_api.js b/lib/mq_client_api.js index a5f0e9b..6260e9c 100644 --- a/lib/mq_client_api.js +++ b/lib/mq_client_api.js @@ -531,6 +531,46 @@ class MQClientAPI extends RemotingClient { } } + async lockBatchMQ(brokerAddr, consumerGroup, clientId, mqSet, timeoutMillis) { + const requestBody = { + consumerGroup, + clientId, + mqSet, + }; + const request = RemotingCommand.createRequestCommand(RequestCode.LOCK_BATCH_MQ, null); + request.body = new Buffer(JSON.stringify(requestBody)); + const response = await this.invoke(brokerAddr, request, timeoutMillis); + switch (response.code) { + case ResponseCode.SUCCESS: + if (response.body) { + const body = JSON2.parse(response.body.toString()); + return body.lockOKMQSet.map(mq => new MessageQueue(mq.topic, mq.brokerName, mq.queueId)); + } + break; + default: + this._defaultHandler(request, response); + break; + } + } + + async unlockBatchMQ(brokerAddr, consumerGroup, clientId, mqSet, timeoutMillis) { + const requestBody = { + consumerGroup, + clientId, + mqSet, + }; + const request = RemotingCommand.createRequestCommand(RequestCode.UNLOCK_BATCH_MQ, null); + request.body = new Buffer(JSON.stringify(requestBody)); + const response = await this.invoke(brokerAddr, request, timeoutMillis); + switch (response.code) { + case ResponseCode.SUCCESS: + return; + default: + this._defaultHandler(request, response); + break; + } + } + // * viewMessage(brokerAddr, phyoffset, timeoutMillis) { // const requestHeader = { // offset: phyoffset, diff --git a/lib/process_queue.js b/lib/process_queue.js index 420255e..3d261c1 100644 --- a/lib/process_queue.js +++ b/lib/process_queue.js @@ -5,6 +5,7 @@ const bsInsert = require('binary-search-insert'); const comparator = (a, b) => a.queueOffset - b.queueOffset; const pullMaxIdleTime = 120000; +const lockMaxLiveTime = 30000; class ProcessQueue extends Base { constructor(options = {}) { @@ -17,6 +18,12 @@ class ProcessQueue extends Base { this.locked = false; this.lastLockTimestamp = Date.now(); + this.consuming = false; + this.orderlyProcessChain = Promise.resolve(); + } + + get lockExpired() { + return (Date.now() - this.lastLockTimestamp) > lockMaxLiveTime; } get maxSpan() { diff --git a/lib/producer/mq_producer.js b/lib/producer/mq_producer.js index 4708414..e3540e2 100644 --- a/lib/producer/mq_producer.js +++ b/lib/producer/mq_producer.js @@ -183,9 +183,10 @@ class MQProducer extends ClientConfig { /** * send message * @param {Message} msg - message object + * @param {Function/String} selector - select method * @return {Object} sendResult */ - async send(msg) { + async send(msg, selector) { await this.ready(); msg.topic = this.formatTopic(msg.topic); @@ -203,6 +204,35 @@ class MQProducer extends ClientConfig { let times = 0; let lastBrokerName; let sendResult; + + if (typeof selector === 'string') { + const shardingKey = selector; + selector = mqs => { + let select = Math.abs(utils.hashCode(shardingKey)); + if (select < 0) { + select = 0; + } + return mqs[select % mqs.length]; + }; + } + if (typeof selector === 'function') { + let messageQueue = null; + try { + // 克隆messgeQueueList以防止被用户破坏 + const messgeQueueList = topicPublishInfo.messageQueueList.map(e => Object.assign({}, e)); + messageQueue = selector(messgeQueueList); + } catch (e) { + throw new Error('select message queue throwed exception.'); + } + if (!messageQueue) { + throw new Error('select message queue return null.', null); + } + // 从用户选择的messageQueue还原为内部的messageQueue, 因为用户给的结果是不可靠的 + messageQueue = topicPublishInfo.messageQueueList.find(mq => messageQueue.key === mq.key); + sendResult = await this.sendKernelImpl(msg, messageQueue); + return sendResult; + } + for (; times < timesTotal && Date.now() - beginTimestamp < maxTimeout; times++) { const messageQueue = topicPublishInfo.selectOneMessageQueue(lastBrokerName); if (!messageQueue) { diff --git a/test/index.test.js b/test/index.test.js index 5eeb2e2..bf841de 100644 --- a/test/index.test.js +++ b/test/index.test.js @@ -147,6 +147,8 @@ describe('test/index.test.js', () => { }); it('should updateProcessQueueTableInRebalance ok', async () => { + mm(consumer, 'consumeOrderly', true); + mm(consumer, 'messageModel', 'CLUSTERING'); await sleep(3000); await consumer.rebalanceByTopic(TOPIC); const size = consumer.processQueueTable.size; @@ -160,7 +162,7 @@ describe('test/index.test.js', () => { await consumer.rebalanceByTopic(TOPIC); assert(consumer.processQueueTable.size === size); - await consumer.updateProcessQueueTableInRebalance(TOPIC, []); + await consumer.updateProcessQueueTableInRebalance(TOPIC, [], true); assert(consumer.processQueueTable.size === 0); await consumer.updateProcessQueueTableInRebalance(MixAll.getRetryTopic(consumer.consumerGroup), []); assert(consumer.processQueueTable.size === 0); @@ -645,4 +647,192 @@ describe('test/index.test.js', () => { assert(consumeTime - produceTime <= delayTime + deviationTime && consumeTime - produceTime >= delayTime - deviationTime); }); }); + + // 顺序消息 + describe('order message', () => { + let consumer; + let producer; + beforeEach(async () => { + consumer = new Consumer(Object.assign({ + httpclient, + isBroadcast: false, + consumeOrderly: true, + maxReconsumeTimes: 2, + }, config)); + await consumer.ready(); + producer = new Producer(Object.assign({ + httpclient, + }, config)); + await producer.ready(); + }); + + afterEach(async () => { + await consumer.close(); + await producer.close(); + mm.restore(); + }); + + it('msgs should be in same queue and follow order', async () => { + await sleep(3000); + + const msg = new Message(TOPIC, // topic + 'TagC', // tag + 'Hello MetaQ !!! ' // body + ); + const selector = Math.random().toString(); + const sendResult = await producer.send(msg, selector); + assert(sendResult && sendResult.msgId); + const sendResult_2 = await producer.send(msg, selector); + assert(sendResult_2 && sendResult_2.msgId); + assert(sendResult.messageQueue.key === sendResult_2.messageQueue.key); + + console.log(sendResult, sendResult_2, '2 results'); + let queueId = null; + await new Promise(r => { + consumer.subscribe(TOPIC, 'TagC', async msg => { + if (queueId === null) { // first msg + queueId = msg.queueId; + } + if (queueId !== null) { // 2nd msg + console.log(msg.queueId); + assert(msg.queueId === queueId); // 2 msgs have same queueId + r(); + } + }); + }); + }); + + it('function as selector', async () => { + await sleep(3000); + + const msg = new Message(TOPIC, // topic + 'TagC', // tag + 'Hello MetaQ !!! ' // body + ); + const shardingKey = Math.random().toString(); + const selector = mqs => { + let select = Math.abs(utils.hashCode(shardingKey)); + if (select < 0) { + select = 0; + } + return mqs[select % mqs.length]; + }; + const sendResult = await producer.send(msg, selector); + assert(sendResult && sendResult.msgId); + const sendResult_2 = await producer.send(msg, selector); + assert(sendResult_2 && sendResult_2.msgId); + assert(sendResult.messageQueue.key === sendResult_2.messageQueue.key); + + console.log(sendResult, sendResult_2, '2 results'); + let queueId = null; + await new Promise(r => { + consumer.subscribe(TOPIC, 'TagC', async msg => { + if (queueId === null) { // first msg + queueId = msg.queueId; + } + if (queueId !== null) { // 2nd msg + console.log(msg.queueId); + assert(msg.queueId === queueId); // 2 msgs have same queueId + r(); + } + }); + }); + }); + + // 顺序消息消费失败 + it('failed msg should be dropped', async () => { + await sleep(3000); + + const msg = new Message(TOPIC, // topic + 'TagC', // tag + 'Hello MetaQ !!! ' // body + ); + const msg_2 = new Message(TOPIC, // topic + 'TagC', // tag + 'Hello MetaQ !!! 2nd' // body + ); + const selector = Math.random().toString(); + const sendResult = await producer.send(msg, selector); + assert(sendResult && sendResult.msgId); + const sendResult_2 = await producer.send(msg_2, selector); + assert(sendResult_2 && sendResult_2.msgId); + + console.log(sendResult, sendResult_2, '2 results'); + let queueId = null; + await new Promise(r => { + consumer.subscribe(TOPIC, 'TagC', async msg => { + if (msg.body.toString() === 'Hello MetaQ !!! ') { // first msg + queueId = msg.queueId; + throw Error('msg processing failed'); + } + if (queueId !== null) { // 2nd msg + console.warn('message receive ------------> ', msg.body.toString()); + if (msg.msgId === sendResult_2.msgId) { + assert(msg.body.toString() === 'Hello MetaQ !!! 2nd'); + r(); + } + } + }); + }); + }); + }); + + // 顺序消息,锁失效逻辑 + describe('order message & lock expired', () => { + let consumer; + let producer; + beforeEach(async () => { + consumer = new Consumer(Object.assign({ + httpclient, + isBroadcast: false, + consumeOrderly: true, + }, config)); + mm(consumer, 'lockAll', () => { + return; + }); + await consumer.ready(); + producer = new Producer(Object.assign({ + httpclient, + }, config)); + await producer.ready(); + }); + + afterEach(async () => { + await consumer.close(); + await producer.close(); + mm.restore(); + }); + + it('should relock when lock expired', async () => { + await sleep(3000); + + const msg = new Message(TOPIC, // topic + 'TagD', // tag + 'Hello MetaQ !!! ' // body + ); + const selector = Math.random().toString(); + const sendResult = await producer.send(msg, selector); + assert(sendResult && sendResult.msgId); + const sendResult_2 = await producer.send(msg, selector); + assert(sendResult_2 && sendResult_2.msgId); + + console.log(sendResult, sendResult_2, '2 results'); + let queueId = null; + await new Promise(r => { + consumer.subscribe(TOPIC, 'TagD', async msg => { + // parallelConsumeLimit为1的时候，第一条消息拖到锁失效 + if (queueId === null) { // first msg + console.log('sleep until lock expired'); + await sleep(30000); + } + if (queueId !== null) { + console.log(msg.queueId); + assert(msg.queueId === queueId); + r(); + } + queueId = msg.queueId; + }); + }); + }); + }); });