日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

大家好，我是君哥。今天來聊一聊 RocketMQ 客戶端消息消費(fèi)失敗，怎么辦？

下面是 RocketMQ 推模式的一段代碼：

public static void main(String[] args) throws InterruptedException, MQClientException {
 Tracer tracer = initTracer();
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
 consumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new ConsumeMessageOpenTracingHookImpl(tracer));
 consumer.subscribe("TopicTest", "*");
 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 consumer.setConsumeTimestamp("20181109221800");
 consumer.registerMessageListener(new MessageListenerConcurrently() {
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
   try{
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
   }catch (Exception e){
    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
   }
   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }
 });
 consumer.start();
}

從這段代碼可以看出，消費(fèi)者消費(fèi)消息后會返回一個消費(fèi)狀態(tài)，那消費(fèi)狀態(tài)有哪些呢？參見類 ConsumeConcurrentlyStatus 中定義：

• 消費(fèi)成功，返回 CONSUME_SUCCESS。
• 消費(fèi)失敗，返回 RECONSUME_LATER。

下面代碼就是返回上面兩個狀態(tài)的邏輯，對于消費(fèi)狀態(tài)，如果返回 null，會給它賦值 RECONSUME_LATER，處理邏輯如下：

//ConsumeRequest 類
public void run() {
 MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
 //省略部分邏輯
 long beginTimestamp = System.currentTimeMillis();
 ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
 try {
  //省略部分邏輯
  status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
 } catch (Throwable e) {}
    //省略部分邏輯
 if (null == status) {
  //省略日志
  status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
 }
    //省略部分邏輯
 if (!processQueue.isDropped()) {
  ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
 } else {}
}

這部分代碼的 UML 類圖如下：

上面代碼中的 processConsumeResult 方法就是消費(fèi)失敗后客戶端的處理邏輯：

public void processConsumeResult(
 final ConsumeConcurrentlyStatus status,
 final ConsumeConcurrentlyContext context,
 final ConsumeRequest consumeRequest
) {
    //ackIndex 初始值是 Integer.MAX_VALUE;
 int ackIndex = context.getAckIndex();
 switch (status) {
  case CONSUME_SUCCESS:
   if (ackIndex >= consumeRequest.getMsgs().size()) {
    ackIndex = consumeRequest.getMsgs().size() - 1;
   }
   //省略部分邏輯
   break;
  case RECONSUME_LATER:
   ackIndex = -1;
   //省略部分邏輯
   break;
  default:
   break;
 }
 switch (this.defaultMQPushConsumer.getMessageModel()) {
  case BROADCASTING:
   //廣播模式下這里只打印日志
   break;
  case CLUSTERING:
   List msgBackFailed = new ArrayList(consumeRequest.getMsgs().size());
   for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
    MessageExt msg = consumeRequest.getMsgs().get(i);
    boolean result = this.sendMessageBack(msg, context);
    if (!result) {
     msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
     msgBackFailed.add(msg);
    }
   }

   if (!msgBackFailed.isEmpty()) {
    consumeRequest.getMsgs().removeAll(msgBackFailed);
    //發(fā)送回 Broker 失敗的消息，5s 后再次消費(fèi)
    this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
   }
   break;
  default:
   break;
 }
     //更新本地保存的偏移量
 long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
 if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
  this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
 }
}

一 、消費(fèi)成功

上面的代碼邏輯中，如果消費(fèi)成功，ackIndex 變量的值就是消息數(shù)量減 1，所以上面的 switch 邏輯是不會執(zhí)行的，因為廣播模式下，只是打印一段日志（沒有其他邏輯），而集群模式下，for 循環(huán)的起始 i 變量已經(jīng)等于消息數(shù)量，循環(huán)里面的代碼不會執(zhí)行。

因此，如果消息消費(fèi)成功，只會走最下面的邏輯，更新本地保存的消息偏移量。

二 、消費(fèi)失敗

ackIndex 變量值等于 -1。

1、廣播模式

在消費(fèi)失敗的情況下，廣播模式的代碼只是打印了一段日志，之后更新了本地保存的消息偏移量，因此我們知道廣播模式消息消費(fèi)失敗后就不會重新消費(fèi)了，相當(dāng)于丟棄了消息。

2、集群模式

從上面代碼的 for 循環(huán)中，會把所有的消息都發(fā)送回 Broker，這樣這批消息還能再次被拉取到進(jìn)行消費(fèi)。

對于發(fā)送給 Broker 失敗的消息，會延遲 5s 后再次消費(fèi)。代碼如下：

private void submitConsumeRequestLater(
 final List msgs,
 final ProcessQueue processQueue,
 final MessageQueue messageQueue
) {
 this.scheduledExecutorService.schedule(new Runnable() {
  @Override
  public void run() {
   ConsumeMessageConcurrentlyService.this.submitConsumeRequest(msgs, processQueue, messageQueue, true);
  }
 }, 5000, TimeUnit.MILLISECONDS);
}

更新本地保存的消息偏移量時，會從消息列表中把發(fā)送回 Broker 失敗的消息先刪除掉。

注意：從上面邏輯可以看到，在拉取到一批消息進(jìn)行消費(fèi)時，只要有一條消息消費(fèi)失敗，這批消息都會進(jìn)行重試，因此消費(fèi)端做好冪等是必要的。

下面再看一下發(fā)送失敗消息給 Broker 的代碼，發(fā)送消息是，請求的 code 碼是 CONSUMER_SEND_MSG_BACK。根據(jù)這個請求碼就能找 Broker 端的處理邏輯。

如果發(fā)送回 Broker 時拋出異常，需要重新發(fā)送一個新的消息，這里有四點需要注意：

• 新消息的 Topic 變成【 %RETRY% + consumerGroup】。
• 新消息的 RETRY_TOPIC 這個屬性賦值為之前的 Topic。
• 新消息的重試次數(shù)屬性加 1；
• 新消息的 DELAY 屬性等于重試次數(shù) + 3。

public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)
 throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
 try {
  this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,
   this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());
 } catch (Exception e) {
  //Topic 變成 %RETRY% + consumerGroup
  Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());
  String originMsgId = MessageAccessor.getOriginMessageId(msg);
  MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
        //RETRY_TOPIC 賦值
  MessageAccessor.putProperty(newMsg, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());
  //重試次數(shù)+1
  MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes() + 1));
  //最大重試次數(shù)
  MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(getMaxReconsumeTimes()));
  //DELAY = 重試次數(shù) + 3
  newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
  this.mQClientFactory.getDefaultMQProducer().send(newMsg);
 } finally {
  msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQPushConsumer.getNamespace()));
 }
}

3、Broker 處理

上面已經(jīng)講過，對于處理失敗的消息，消費(fèi)端會發(fā)送回 Broker，不過這里有一點需要注意，發(fā)送回 Broker 時，消息的 Topic 變成【"%RETRY%" + namespace + "%" + 原始 topic】，封裝邏輯在源碼 ClientConfig.withNamespace。

根據(jù)請求碼 CONSUMER_SEND_MSG_BACK 可以定位到 Broker 的處理邏輯在類 SendMessageProcessor，方法 asyncConsumerSendMsgBack。

（1）進(jìn)死信隊列

如果重試次數(shù)超過了最大重試次數(shù)（默認(rèn) 16 次），或者 delayLevel 值小于0，則消息進(jìn)死信隊列，死信隊列的 Topic 為【"%DLQ%" + 消費(fèi)組】，代碼如下：

//asyncConsumerSendMsgBack 方法
if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
 || delayLevel < 0) {
 newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
 queueIdInt = ThreadLocalRandom.current().nextInt(99999999) % DLQ_NUMS_PER_GROUP;
 topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
   DLQ_NUMS_PER_GROUP,
   PermName.PERM_WRITE | PermName.PERM_READ, 0); 
 msgExt.setDelayTimeLevel(0);
}

（2）發(fā)送 CommitLog

如果延遲級別（DELAY）等于 0，則延遲級別就等于重試次數(shù)加 3。

有個地方需要注意，發(fā)送到延遲隊列的消息重新進(jìn)行了封裝，封裝這個消息用的并不是客戶端發(fā)來的那個消息，而是從 CommitLog 中根據(jù)偏移量查找的，代碼如下：

MessageExt msgExt = this.brokerController.getMessageStore().lookMessageByOffset(requestHeader.getOffset());
if (null == msgExt) {
 response.setCode(ResponseCode.SYSTEM_ERROR);
 response.setRemark("look message by offset failed, " + requestHeader.getOffset());
 return CompletableFuture.completedFuture(response);
}

如果查詢失敗，就會給客戶端返回系統(tǒng)錯誤。

這里有個重要的細(xì)節(jié)，這個消息寫入 CommitLog 時，會判斷 DELAY 是否大于 0，如果大于 0，就會修改 Topic。代碼如下：

//CommitLog 類 asyncPutMessage 方法
if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
  || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
 // Delay Delivery
 if (msg.getDelayTimeLevel() > 0) {
  if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
      //從源碼看，這里最大值是18
   msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
  }
       
  topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
  //queueId = delayLevel - 1
  int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
  // Backup real topic, queueId
  MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
  MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
  msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
  msg.setTopic(topic);
  msg.setQueueId(queueId);
 }
}

這里把 Topic 修改為 SCHEDULE_TOPIC_XXXX，供延時隊列來調(diào)度。進(jìn)入延時隊列后，延時隊列會按照下面的時間進(jìn)行調(diào)度：

private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

上面代碼可以看到，延時消息的調(diào)度有 18 個等級，最小的 1s，最大的 2h。而從下面的代碼我們可以看到，調(diào)度使用第三個等級開始的：

if (0 == delayLevel) {
 delayLevel = 3 + msgExt.getReconsumeTimes();
}
msgExt.setDelayTimeLevel(delayLevel);

（3）延時隊列

延時隊列的代碼邏輯在類 ScheduleMessageService，這里的 start 方法觸發(fā)延時隊列的調(diào)度，而 start 方法的業(yè)務(wù)入口在 BrokerStartup 的初始化。

首先，會計算出每個延時等級對應(yīng)的延時時間(處理到 ms 級別)，放到 delayLevelTable，它是一個 ConcurrentHashMap，然后創(chuàng)建一個核心線程數(shù)等于 18 的定時線程池，依次對每個級別的延時進(jìn)行調(diào)度。這個任務(wù)啟動后，會每 100ms 執(zhí)行一次。代碼如下：

public void start() {
 if (started.compareAndSet(false, true)) {
  this.load();
  this.deliverExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));
  //省略異步
  for (Map.Entry entry : this.delayLevelTable.entrySet()) {
   Integer level = entry.getKey();
   Long timeDelay = entry.getValue();
   Long offset = this.offsetTable.get(level);
   if (null == offset) {
    offset = 0L;
   }

   if (timeDelay != null) {
    //省略異步
    this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
   }
  }
  //省略其他邏輯
 }
}

調(diào)度邏輯中，首先根據(jù) Topic 和 queueId 找到對應(yīng)的消費(fèi)隊列，然后從里面連續(xù)讀取消息：

public void executeOnTimeup() {
 ConsumeQueue cq =
  ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
   delayLevel2QueueId(delayLevel));
    //省略空處理
 SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
 //省略空處理
 long nextOffset = this.offset;
 try {
  int i = 0;
  ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
  //CQ_STORE_UNIT_SIZE = 20,因為 ConsumeQueue 中一個元素占 20 字節(jié)
  for (; i < bufferCQ.getSize() && isStarted(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
      //offset占8個字節(jié)
   long offsetPy = bufferCQ.getByteBuffer().getLong();
   //消息大小占4個字節(jié)
   int sizePy = bufferCQ.getByteBuffer().getInt();
   //ConsumeQueue中tagsCode是一個投遞時間點
   long tagsCode = bufferCQ.getByteBuffer().getLong();

   if (cq.isExtAddr(tagsCode)) {
    if (cq.getExt(tagsCode, cqExtUnit)) {
     tagsCode = cqExtUnit.getTagsCode();
    } else {
     //can't find ext content.So re compute tags code.
     long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
     tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
    }
   }

   long now = System.currentTimeMillis();
   long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
   nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
   long countdown = deliverTimestamp - now;
   if (countdown > 0) {
       //時間未到，等待下次調(diào)度
    this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
    return;
   }
   MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);
   MessageExtBrokerInner msgInner = ScheduleMessageService.this.messageTimeup(msgExt);
   //省略事務(wù)消息
   boolean deliverSuc;
   //同步異步都有，只保留同步代碼
   deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), nextOffset, offsetPy, sizePy);
  }

  nextOffset = this.offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
 } catch (Exception e) {
 } finally {
  bufferCQ.release();
 }
    //DELAY_FOR_A_WHILE是 100ms
 this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
}

因為 messageTimeup 方法使用了原始的 Topic 和 QueueId 新建了消息，所以上面的 syncDeliver 方式是將消息重新投遞到原始的隊列中，這樣消費(fèi)者可以再次拉取到這條消息進(jìn)行消費(fèi)。注意：上面 ConsumeQueue 的 tagsCode 是一個時間點，很容易誤解為是 tag 的 hashCode，MessageQueue 的存儲元素中最后 8 字節(jié)確實是 tag 的 hashCode。

三、總結(jié)

消費(fèi)者消費(fèi)失敗后，會把消費(fèi)發(fā)回給 Broker 進(jìn)行處理。下圖是客戶端處理流程：

Broker 收到消息后，會把消息重新發(fā)送到 CommitLog，發(fā)送到 CommitLog 之前，首先會修改 Topic 為 SCHEDULE_TOPIC_XXXX，這樣就發(fā)送到了延時隊列，延時隊列再根據(jù)延時級別把消息投遞到原始的隊列，這樣消費(fèi)者就能再次拉取到。流程如下圖：

從流程來看，消費(fèi)者批量拉取消息，如果部分消息消費(fèi)失敗，那就會整批全部重試。所以做好冪等是必要的。

名稱欄目：阿里二面：RocketMQ消費(fèi)失敗了，怎么處理？
當(dāng)前URL：http://m.5511xx.com/article/ccdiseg.html