feat: support consume back #56

denghongcai · 2018-11-21T03:46:00Z

#34 support RECONSUME_LATER status, send back to broker.

codecov · 2018-11-21T03:53:01Z

Codecov Report

Merging #56 into master will increase coverage by 0.93%.
The diff coverage is 91.86%.

@@            Coverage Diff             @@
##           master      #56      +/-   ##
==========================================
+ Coverage   90.59%   91.52%   +0.93%     
==========================================
  Files          35       35              
  Lines        1723     1806      +83     
==========================================
+ Hits         1561     1653      +92     
+ Misses        162      153       -9

Impacted Files	Coverage Δ
lib/mq_client.js	`88.27% <ø> (+0.34%)`	⬆️
lib/producer/mq_producer.js	`95.2% <100%> (+4.04%)`	⬆️
lib/message/message_const.js	`100% <100%> (ø)`	⬆️
lib/mix_all.js	`100% <100%> (+11.11%)`	⬆️
lib/message/message.js	`100% <100%> (ø)`	⬆️
lib/mq_client_api.js	`76.74% <71.42%> (+2.7%)`	⬆️
lib/utils/index.js	`96.55% <80%> (-3.45%)`	⬇️
lib/consumer/mq_push_consumer.js	`93.6% <92.3%> (+0.77%)`	⬆️
... and 3 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f1258c2...beb756c. Read the comment docs.

gxcsoccer · 2018-12-10T08:55:02Z

这个我晚点仔细看下

gxcsoccer · 2018-12-10T18:32:37Z

lib/producer/mq_producer.js

@@ -78,6 +78,7 @@ class MQProducer extends ClientConfig {
   */
  async init() {
    // this._topicPublishInfoTable.set(this.createTopicKey, new TopicPublishInfo());
+    await MQProducer.getDefaultProducer(this.options);


这里为啥要 getDefaultProducer ?

初始化一下，CLIENT_INNER_PRODUCER 的 Producer 是默认就要有的

gxcsoccer · 2018-12-10T18:43:00Z

lib/consumer/mq_push_consumer.js

+    })();
+  }
+
+  async _consumeMessageLoop(topic, needFilter, tagsSet, handler, subExpression) {


这段现在好复杂

gxcsoccer · 2018-12-10T18:43:45Z

consume back 和并发消费，其实可以分开 pr 的，改动都比较大

gxcsoccer · 2018-12-10T18:46:23Z

lib/consumer/mq_push_consumer.js

+              if (!msg.tags || !needFilter || tagsSet.includes(msg.tags)) {
+                try {
+                  if (msg.reconsumeTimes < this.options.maxReconsumeTimes) {
+                    await handler(msg, mq, pq);


这里还是一条一条的串行的吧？

改了，你再看看

gxcsoccer · 2018-12-12T08:35:46Z

lib/consumer/mq_push_consumer.js

+      this.subscriptions.set(retryTopic, {
+        subscriptionData: retrySubscriptionData,
+      });
+    }


这个可以在最开始就初始化好，不用每次都做

最开始初始化，没有注册过 handler 的情况下会导致找不到 handler

这一段好像和 handler 无关的吧

应该是在初始化的时候就监听统一的 retryTopic，retryTopic 的 handler 比较特殊，就是从 this.subscriptions 里面找原始 topic 的 handler 吐给他

gxcsoccer · 2018-12-12T08:57:01Z

lib/consumer/mq_push_consumer.js

+          this.logger.error('[MQPushConsumer] send reconsume message failed, fall to local retry, msgId: %s', msg.msgId);
+          // 重试消息发送失败，本地重试
+          msg.reconsumeTimes++;
+          await this._sleep(5000);


这里如果真的失败的，会导致这批消息重新消费

这里还不如一只让他定时重发，直到成功。

因为直接跑错的效果也是类似的，还会导致其他消息重复消费

这里好像还没改

把上面的循环判断去掉了，永远重试

gxcsoccer · 2018-12-12T09:09:46Z

lib/consumer/mq_push_consumer.js

+
+        if (this.messageModel === MessageModel.CLUSTERING) {
+          const retryTopic = MixAll.getRetryTopic(this.consumerGroup);
+          loops.push((async () => {


这样会重复注册 retryTopic 的消费逻辑吧？

gxcsoccer · 2018-12-12T13:23:04Z

lib/consumer/mq_push_consumer.js

+
+  async consumeSingleMsg(handler, msg, mq, pq) {
+    try {
+      if (msg.reconsumeTimes < this.options.maxReconsumeTimes) {


如果超过 maxResconsumeTimes 应该有个报错吧

这个判断直接去掉了

gxcsoccer · 2018-12-12T13:24:13Z

lib/consumer/mq_push_consumer.js

+            // 并发消费任务
+            const consumeTasks = [];
+            for (const msg of msgs) {
+              utils.resetRetryTopic(msg, this.consumerGroup);


这个应该放到 retryTopic 的订阅逻辑里吧？

恩，改了，忘记删了

gxcsoccer · 2018-12-12T13:26:59Z

lib/consumer/mq_push_consumer.js

+      : msg.storeHost;
+    const thatConsumerGroup = consumerGroup ? consumerGroup : this.consumerGroup;
+    try {
+      await this._mqClient.consumerSendMessageBack(


consumerSendMessageBack 和 send retry message 有什么区别？

consumerSendMessageBack 不需要发消息体，send retry message 需要发

gxcsoccer · 2018-12-14T09:04:47Z

lib/consumer/mq_push_consumer.js

+            this.emit('error', err);
+            this.logger.error('[MQPushConsumer] send reconsume message failed, fallback to local retry, msgId: %s', msg.msgId);
+            // 重试消息发送失败，本地重试
+            msg.reconsumeTimes++;


这个是不是应该挪到上个 catch 里？

gxcsoccer · 2018-12-14T09:05:16Z

其他 +1

gxcsoccer · 2018-12-14T11:04:14Z

3.3.0

denghongcai force-pushed the feat-support-cosume-back branch from 64b52d6 to 4046d5d Compare November 21, 2018 03:53

denghongcai requested a review from gxcsoccer December 10, 2018 06:14

denghongcai changed the title ~~[WIP] feat: support consume back~~ feat: support consume back Dec 10, 2018

denghongcai force-pushed the feat-support-cosume-back branch 3 times, most recently from dc0796c to 11039f6 Compare December 10, 2018 06:27

denghongcai mentioned this pull request Dec 10, 2018

并发性能 #23

Closed

denghongcai force-pushed the feat-support-cosume-back branch 2 times, most recently from 6a2e213 to 13536e3 Compare December 10, 2018 07:03

gxcsoccer reviewed Dec 10, 2018

View reviewed changes

denghongcai force-pushed the feat-support-cosume-back branch 2 times, most recently from 37515ec to ad31907 Compare December 12, 2018 07:28

gxcsoccer reviewed Dec 12, 2018

View reviewed changes

denghongcai force-pushed the feat-support-cosume-back branch 3 times, most recently from 9d98427 to a65d043 Compare December 12, 2018 10:21

gxcsoccer reviewed Dec 12, 2018

View reviewed changes

denghongcai force-pushed the feat-support-cosume-back branch 3 times, most recently from c8b0c7f to a12360d Compare December 14, 2018 07:22

gxcsoccer reviewed Dec 14, 2018

View reviewed changes

feat: support consume back

beb756c

denghongcai force-pushed the feat-support-cosume-back branch from a12360d to beb756c Compare December 14, 2018 09:15

gxcsoccer merged commit 9acff9b into master Dec 14, 2018

gxcsoccer deleted the feat-support-cosume-back branch December 14, 2018 10:38

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support consume back #56

feat: support consume back #56

denghongcai commented Nov 21, 2018 •

edited

Loading

codecov bot commented Nov 21, 2018 •

edited

Loading

gxcsoccer commented Dec 10, 2018

gxcsoccer Dec 10, 2018

denghongcai Dec 11, 2018

gxcsoccer Dec 10, 2018

gxcsoccer commented Dec 10, 2018

gxcsoccer Dec 10, 2018

denghongcai Dec 11, 2018

gxcsoccer Dec 12, 2018 •

edited

Loading

denghongcai Dec 12, 2018

gxcsoccer Dec 12, 2018

gxcsoccer Dec 12, 2018

gxcsoccer Dec 12, 2018

gxcsoccer Dec 12, 2018

gxcsoccer Dec 12, 2018

denghongcai Dec 13, 2018

gxcsoccer Dec 12, 2018

gxcsoccer Dec 12, 2018

denghongcai Dec 13, 2018 •

edited

Loading

gxcsoccer Dec 12, 2018

denghongcai Dec 13, 2018

gxcsoccer Dec 12, 2018

denghongcai Dec 13, 2018

gxcsoccer Dec 14, 2018

gxcsoccer commented Dec 14, 2018

gxcsoccer commented Dec 14, 2018

feat: support consume back #56

feat: support consume back #56

Conversation

denghongcai commented Nov 21, 2018 • edited Loading

codecov bot commented Nov 21, 2018 • edited Loading

Codecov Report

gxcsoccer commented Dec 10, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gxcsoccer commented Dec 10, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gxcsoccer Dec 12, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

denghongcai Dec 13, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gxcsoccer commented Dec 14, 2018

gxcsoccer commented Dec 14, 2018

denghongcai commented Nov 21, 2018 •

edited

Loading

codecov bot commented Nov 21, 2018 •

edited

Loading

gxcsoccer Dec 12, 2018 •

edited

Loading

denghongcai Dec 13, 2018 •

edited

Loading