2. 程式人生 > >RocketMQ——ack機制保證消費成功

RocketMQ——ack機制保證消費成功

阿新 • • 發佈：2019-01-26

ACK簡介

在實際使用RocketMQ的時候我們並不能保證每次傳送的訊息都剛好能被消費者一次性正常消費成功，可能會存在需要多次消費才能成功或者一直消費失敗的情況，那作為傳送者該做如何處理呢？

RocketMQ提供了ack機制，以保證訊息能夠被正常消費。傳送者為了保證訊息肯定消費成功，只有使用方明確表示消費成功，RocketMQ才會認為訊息消費成功。中途斷電，丟擲異常等都不會認為成功——即都會重新投遞。

DEMO

當然，如果消費者不告知傳送者我這邊消費資訊異常，那麼傳送者是不會知道的，所以消費者在設定監聽的時候需要給個回撥，具體程式碼如下：

consumer.registerMessageListener(new MessageListenerConcurrently() {  
  
             @Override  
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                             ConsumeConcurrentlyContext context) {  
                 for (MessageExt messageExt : msgs) {    
                    String messageBody = new String(messageExt.getBody());    
                    System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                    		 new Date())+"消費響應：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//輸出訊息內容    
                 }    
                   
                 //返回消費狀態  
                 //CONSUME_SUCCESS 消費成功  
                 //RECONSUME_LATER 消費失敗，需要稍後重新消費  
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;  
             }  
         });

業務實現消費回撥的時候，當且僅當此回撥函式返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS，RocketMQ才會認為這批訊息（預設是1條）是消費完成的。如果這時候訊息消費失敗，例如資料庫異常，餘額不足扣款失敗等一切業務認為訊息需要重試的場景，只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER，RocketMQ就會認為這批訊息消費失敗了。

為了保證訊息是肯定被至少消費成功一次，RocketMQ會把這批訊息重發回Broker（topic不是原topic而是這個消費租的RETRY topic），在延遲的某個時間點（預設是10秒，業務可設定）後，再次投遞到這個ConsumerGroup。而如果一直這樣重複消費都持續失敗到一定次數（預設16次），就會投遞到DLQ死信佇列。應用可以監控死信佇列來做人工干預。

修改重試時間

重試的時間預設如下，這個可以通過檢視broker的日誌，

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

當然，這個重複的時間間隔是可以在配置檔案內設定的，由於我這邊配置的雙master模式，所以在128伺服器的broker-a.properties和129的broker-b.properties中分別配置，如下圖，設定好後務必將之前的資料清理

messageDelayLevel = 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s 5s

重試訊息的處理

一般情況下我們在實際生產中是不需要重試16次，這樣既浪費時間又浪費效能，理論上當嘗試重複次數達到我們想要的結果時如果還是消費失敗，那麼我們需要將對應的訊息進行記錄，並且結束重複嘗試。

package com.gwd.rocketmq;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

/** 
* @FileName Consumer.java
* @Description:
* @author gu.weidong
* @version V1.0
* @createtime 2018年6月25日 上午9:49:39 
* 修改歷史：
* 時間           作者          版本        描述
*====================================================  
*
*/
public class Consumer {
	public static void main(String[] args) throws MQClientException {  
        
        //宣告並初始化一個consumer  
         //需要一個consumer group名字作為構造方法的引數，這裡為consumer1  
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");  
  
         //同樣也要設定NameServer地址  
         consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");  
         //這裡設定的是一個consumer的消費策略  
         //CONSUME_FROM_LAST_OFFSET 預設策略，從該佇列最尾開始消費，即跳過歷史訊息  
         //CONSUME_FROM_FIRST_OFFSET 從佇列最開始開始消費，即歷史訊息（還儲存在broker的）全部消費一遍  
         //CONSUME_FROM_TIMESTAMP 從某個時間點開始消費，和setConsumeTimestamp()配合使用，預設是半個小時以前  
         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
         //設定consumer所訂閱的Topic和Tag，*代表全部的Tag  
         consumer.subscribe("TopicTest", "*");  
        //設定一個Listener，主要進行訊息的邏輯處理  
         consumer.registerMessageListener(new MessageListenerConcurrently() {  
             @Override  
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                             ConsumeConcurrentlyContext context) {  
                 for (MessageExt messageExt : msgs) {    
	                 if(messageExt.getReconsumeTimes()==3) {
	                	 //可以將對應的資料儲存到資料庫，以便人工干預
	                	 System.out.println(messageExt.getMsgId()+","+messageExt.getBody());
	                	 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	                 }
                 }       
                 //返回消費狀態  
                 //CONSUME_SUCCESS 消費成功  
                 //RECONSUME_LATER 消費失敗，需要稍後重新消費  
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;  
             }  
         });  
         //呼叫start()方法啟動consumer  
         consumer.start();  
         System.out.println("Consumer Started.");  
     }  
}

所以任何異常都要捕獲返回ConsumeConcurrentlyStatus.RECONSUME_LATER,rocketmq會放到重試佇列,這個重試TOPIC的名字是%RETRY%+consumergroup的名字,如下圖：

注意點

1.如果業務的回撥沒有處理好而丟擲異常，會認為是消費失敗當ConsumeConcurrentlyStatus.RECONSUME_LATER處理。

2.當使用順序消費的回撥MessageListenerOrderly時，由於順序消費是要前者消費成功才能繼續消費，所以沒有ConsumeConcurrentlyStatus.RECONSUME_LATER的這個狀態，只有ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT來暫停佇列的其餘消費，直到原訊息不斷重試成功為止才能繼續消費。

測試案例

（1）producer

package com.gwd.rocketmq;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;

/** 
* @FileName Producer.java
* @Description:
* @author gu.weidong
* @version V1.0
* @createtime 2018年6月25日 上午9:48:37 
* 修改歷史：
* 時間           作者          版本        描述
*====================================================  
*
*/
public class Producer {
	public static void main(String[] args) throws MQClientException, InterruptedException {  
        //宣告並初始化一個producer  
        //需要一個producer group名字作為構造方法的引數，這裡為producer1  
        DefaultMQProducer producer = new DefaultMQProducer("producer1");  
          
        //設定NameServer地址,此處應改為實際NameServer地址，多個地址之間用；分隔  
        //NameServer的地址必須有，但是也可以通過環境變數的方式設定，不一定非得寫死在程式碼裡  
        producer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");  
   //     producer.setVipChannelEnabled(false);//3.2。6的版本沒有該設定，在更新或者最新的版本中務必將其設定為false，否則會有問題  
        //呼叫start()方法啟動一個producer例項  
        producer.start();  
  
        //傳送10條訊息到Topic為TopicTest，tag為TagA，訊息內容為“Hello RocketMQ”拼接上i的值  
        for (int i = 0; i < 5; i++) {  
            try {  
                Message msg = new Message("TopicTest",// topic  
                        "TagA",// tag  
                        ("Hello RocketMQ " + i).getBytes("utf-8")// body  
                );          
                //呼叫producer的send()方法傳送訊息  
                //這裡呼叫的是同步的方式，所以會有返回結果  
                SendResult sendResult = producer.send(msg);  
                //列印返回結果，可以看到訊息傳送的狀態以及一些相關資訊  
                System.out.println(sendResult);  
            } catch (Exception e) {  
                e.printStackTrace();  
                Thread.sleep(1000);  
            }  
        }  
  
        //傳送完訊息之後，呼叫shutdown()方法關閉producer  
        producer.shutdown();  
    }  
}

（2）Consumer

package com.gwd.rocketmq;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

/** 
* @FileName Consumer.java
* @Description:
* @author gu.weidong
* @version V1.0
* @createtime 2018年6月25日 上午9:49:39 
* 修改歷史：
* 時間           作者          版本        描述
*====================================================  
*
*/
public class Consumer {
	public static void main(String[] args) throws MQClientException {  
        
        //宣告並初始化一個consumer  
         //需要一個consumer group名字作為構造方法的引數，這裡為consumer1  
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");  
  
         //同樣也要設定NameServer地址  
         consumer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");  
         //這裡設定的是一個consumer的消費策略  
         //CONSUME_FROM_LAST_OFFSET 預設策略，從該佇列最尾開始消費，即跳過歷史訊息  
         //CONSUME_FROM_FIRST_OFFSET 從佇列最開始開始消費，即歷史訊息（還儲存在broker的）全部消費一遍  
         //CONSUME_FROM_TIMESTAMP 從某個時間點開始消費，和setConsumeTimestamp()配合使用，預設是半個小時以前  
         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
  
         //設定consumer所訂閱的Topic和Tag，*代表全部的Tag  
         consumer.subscribe("TopicTest", "*");  
  
         //設定一個Listener，主要進行訊息的邏輯處理  
         consumer.registerMessageListener(new MessageListenerConcurrently() {  
             @Override  
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                             ConsumeConcurrentlyContext context) {  
                 for (MessageExt messageExt : msgs) {    
                    String messageBody = new String(messageExt.getBody());    
                     System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                    		 new Date())+"消費響應：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//輸出訊息內容    
                 }    
                 //返回消費狀態  
                 //CONSUME_SUCCESS 消費成功  
                 //RECONSUME_LATER 消費失敗，需要稍後重新消費  
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;  
             }  
         });  
         //呼叫start()方法啟動consumer  
         consumer.start();  
         System.out.println("Consumer Started.");  
     }  
}

（3）測試結果如下

相關推薦

RocketMQ——ack機制保證消費成功

ACK簡介 在實際使用RocketMQ的時候我們並不能保證每次傳送的訊息都剛好能被消費者一次性正常消費成功，可能會存在需要多次消費才能成功或者一直消費失敗的情況，那作為傳送者該做如何處理呢？ RocketMQ提供了ack機制，以保證訊息能夠被正常消費。傳送者為了保證訊息肯

RocketMQ原理（4）——訊息ACK機制及消費進度管理

https://zhuanlan.zhihu.com/p/25265380 consumer的每個例項是靠佇列分配來決定如何消費訊息的。那麼消費進度具體是如何管理的，又是如何保證訊息成功消費的（RocketMQ有保證訊息肯定消費成功的特性（失敗則重試）？ 本文將詳細解

原始碼分析RocketMQ之訊息ACK機制（消費進度）

首先簡要闡述一下訊息消費進度首先消費者訂閱訊息消費佇列（MessageQueue),當生產者將訊息負載傳送到MessageQueue中時，消費訂閱者開始消費訊息，訊息消費過程中，為了避免重複消費，需要一

activemq 應答ack機制與消費訊息過程分析

主要關注有以下兩點： 如何獲取訊息？ 訊息如何應答？ ActiveMQMessageConsumer.receive 這個方法是同步阻塞獲取訊息的方法，嘗試從原始碼分析其工作原理。 @Override public Message re

rabbitmq結合springboot實現ACK機制的消費機制

###1.RabbitMQ介紹 RabbitMQ是實現AMQP（高階訊息佇列協議）的訊息中介軟體的一種，最初起源於金融系統，用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。RabbitMQ主要是為了實現系統之間的雙向解耦而實現的。

kafkaspot在ack機制下如何保證記憶體不溢

http://www.cnblogs.com/intsmaze/p/5947078.html storm框架中的kafkaspout類實現的是BaseRichSpout，它裡面已經重寫了fail和ack方法，所以我們的bolt必須實現ack機制，就可以保證訊息的重新發送；如

【Azure Service Bus】 Service Bus如何確保訊息傳送成功，傳送端是否有Ack機制 

問題描述 Service Bus如何確保訊息傳送成功，傳送端是否有Ack機制(是否有回撥API告訴傳送端，服務端已經收到訊息)？根據對.NET傳送Service Bus訊息程式碼的分析，傳送方法queueClient.SendAsync(message)並沒有返回值，所以無法知道傳送訊息是否成功。 &nb

RocketMQ-quickstart（批量消費）

shu when 多重 代碼 color 最大 except 亂序 throws 一、專業術語 Producer 　　　　消費生產者，負責產生消息，一般由業務系統負責產生消息 Consumer 　　　　消息消費者，負責消費消息，一般是後臺系統負責異步消費

RocketMQ部分數據消費不了問題排查

慢慢 ID 均衡 結果 校驗 實例 .com top 一次 問題現象 今天忽然收到RocketMQ預警信息如下： 提醒有部分數據沒有消費，產生堆積情況。 打開RocketMq-Console-Ng查看如下圖形式： 備註：第一反應是Consumer Group內訂閱了

Linux下TCP延遲重慶時 時 彩源碼下載確認(Delayed Ack)機制導致的時延問題分析

googl 分析 最終 end gin 沒有 奇怪 營銷 大於 重慶時 時 彩下載聯系方式：QQ：2747044651 網址在實際測試中發現，當N大於等於3的情況，第2秒之後，每次第三個recv調用，總會阻塞40毫秒左右，但在分析Server端日誌時，發現所有請求在Serv

針對SpringBoot專案中的引數傳遞使用RSA加密機制保證安全性進行封裝成工具

spring-boot-starter-security 1 概述 針對SpringBoot專案中的引數傳遞使用RSA加密機制保證安全性進行封裝成工具，實現自動加密返回資料、解密傳入資料並對映成json 2 使用  這裡的使用就比較簡單了。 (1)將工具匯入專案中。 (

ActiveMQ訊息傳送機制以及ACK機制詳解

AcitveMQ是作為一種訊息儲存和分發元件，涉及到client與broker端資料互動的方方面面，它不僅要擔保訊息的儲存安全性，還要提供額外的手段來確保訊息的分發是可靠的。   一. ActiveMQ訊息傳送機制     Producer客戶端使用來發送訊息的， Cons

RabbitMq + Spring 實現ACK機制

概念性解讀(Ack的靈活)          首先啊，有的人不是太理解這個Ack是什麼，講的接地氣一點，其實就是一個通知，怎麼說呢，當我監聽消費者，正常情況下，不會出異常，但是如果是出現了異常，甚至是沒有獲取的異常，那是不是這條資料就會作廢，但是我們肯定不希望這樣的情況

RocketMQ探祕——客戶端消費

一、客戶端元件總覽             每個客戶端有一個管理者（MQClientManager），是單例，負責生產MQInstance物件，MQClientManager通過id將多個MQInstance物件放在一個列表中。MQClientManager的id命名規範是

大資料Strom流式處理的ACK機制

package com.neusoft.storm.ack; import java.util.Map; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyCont

kafka四 kafka producer的ack機制和kafka叢集

一、前幾節講了簡單的資料傳送接收，忘了說一個點，就是producer傳送資料的時候怎麼保證資料成功傳送到kafka伺服器上。 org.apache.kafka.clients.producer.Producer的send（）方法有三個過載，其中一個如下： producer

storm中的ack機制

我們知道storm一個很重要的特性是它能夠保證你發出的每條訊息都會被完整處理， 完整處理的意思是指： 一個tuple被完全處理的意思是： 這個tuple以及由這個tuple所導致的所有的tuple都被成功處理。而一個tuple會被認為處理失敗瞭如果這個訊息在

RocketMQ原理學習--消費者消費訊息

        在之前的一篇部落格《RocketMQ原理學習--訊息型別》中我們有介紹過RocketMQ的訊息型別，這篇部落格我們簡單介紹一下RocketMQ消費者是如何消費訊息的。 一、Pull or Push     &nb

【Streaming】我在拓撲中使用了Ack機制，為什麼在Storm UI上有大量Failed資料？

提問： 1. 在Storm UI上，有大量Failed資料，且往往是一旦開始有Fail資料，則Fail資料越積越多 2. 整體事件端到端延遲很大 分析： 當spout接收到大量資料，而後端bolt處理較慢，如果spout傳送的事件在超時時間(topology.mes

Storm的BaseBasicBolt原始碼解析ack機制

http://www.cnblogs.com/intsmaze/p/5924873.html 我們在學習ack機制的時候，我們知道Storm的Bolt有BaseBasicBolt和BaseRichBolt。在BaseBasicBolt中，BasicOutputCollect