快捷導(dǎo)航

基于rocketmq的有序消費(fèi)模式和并發(fā)消費(fèi)模式的區(qū)別說明

更新時間：2021年06月22日 10:18:59 作者：從心歸零

這篇文章主要介紹了基于rocketmq的有序消費(fèi)模式和并發(fā)消費(fèi)模式的區(qū)別說明，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

rocketmq消費(fèi)者注冊監(jiān)聽有兩種模式

有序消費(fèi)MessageListenerOrderly和并發(fā)消費(fèi)MessageListenerConcurrently，這兩種模式返回值不同。

MessageListenerOrderly

正確消費(fèi)返回

ConsumeOrderlyStatus.SUCCESS

稍后消費(fèi)返回

ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT

MessageListenerConcurrently

正確消費(fèi)返回

ConsumeConcurrentlyStatus.CONSUME_SUCCESS

稍后消費(fèi)返回

ConsumeConcurrentlyStatus.RECONSUME_LATER

顧名思義，有序消費(fèi)模式是按照消息的順序進(jìn)行消費(fèi)，但是除此之外，在實(shí)踐過程中我發(fā)現(xiàn)和并發(fā)消費(fèi)模式還有很大的區(qū)別的。

第一，速度，下面我打算用實(shí)驗(yàn)來探究一下。

使用mq發(fā)送消息，消費(fèi)者使用有序消費(fèi)模式消費(fèi)，具體的業(yè)務(wù)是阻塞100ms

Long totalTime = 0L;
Date date1 = null;
Date date2 = new Date();
new MessageListenerOrderly() { 
	@Override
	public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
			ConsumeOrderlyContext context) {
        logger.info("==========CONSUME_START===========");  
		logger.info(Thread.currentThread().getName()  
                            + " Receive New Messages: " + msgs.size());  
        try {
        	if(date1 == null)
        		date1 = new Date();//在第一次消費(fèi)時初始化
        	Thread.sleep(100);
       		logger.info("total:"+(++total));
        	date2 = new Date();
       		totalTime = (date2.getTime() - date1.getTime());
       		logger.info("totalTime:"+totalTime);
            logger.info("==========CONSUME_SUCCESS===========");  
            return ConsumeOrderlyStatus.SUCCESS;  
        }catch (Exception e) {
            logger.info("==========RECONSUME_LATER===========");  
            logger.error(e.getMessage(),e);
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
	}
}

消費(fèi)100條消息

速度挺快的，為了讓結(jié)果更準(zhǔn)確，將消息加到1000條

消費(fèi)1000條消息

可以看到每一條消息平均耗時25ms，然而業(yè)務(wù)是阻塞100ms，這說明有序消費(fèi)模式和同步消費(fèi)可能并不是一回事，那如果不阻塞代碼我們再來看一下結(jié)果

不阻塞過后速度明顯提高了，那么我阻塞300ms會怎么樣呢？

時間相比阻塞100ms多了2倍

接下來我們測試并發(fā)消費(fèi)模式

Long totalTime = 0L;
Date date1 = null;
Date date2 = new Date();
new MessageListenerConcurrently() {
    public ConsumeConcurrentlyStatus consumeMessage(  
                       List< MessageExt > msgs, ConsumeConcurrentlyContext context) {  
 
    		logger.info(Thread.currentThread().getName()  
                                 + " Receive New Messages: " + msgs.size()); 
    		try {
    			if(date1 == null)
    				date1 = new Date();
            	Thread.sleep(100);
           		logger.info("total:"+(++total));
           		date2 = new Date();
           		totalTime = (date2.getTime() - date1.getTime());
           		logger.info("totalTime:"+totalTime);
                logger.info("==========CONSUME_SUCCESS===========");  
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
            } catch (Exception e) {
                logger.info("==========RECONSUME_LATER===========");  
                logger.error(e.getMessage(),e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
    }  
}

基于上次的經(jīng)驗(yàn)，同樣測試三種情況，消費(fèi)1000條不阻塞，消費(fèi)1000條阻塞100ms，消費(fèi)1000條阻塞300ms

消費(fèi)1000條不阻塞的情況

和有序消費(fèi)模式差不多，快個一兩秒。

消費(fèi)1000條阻塞100ms

竟然比不阻塞的情況更快，可能是誤差把

消費(fèi)1000條阻塞300ms

速度稍慢，但是還是比有序消費(fèi)快得多。

結(jié)論是并發(fā)消費(fèi)的消費(fèi)速度要比有序消費(fèi)更快。

另一個區(qū)別是消費(fèi)失敗時的處理不同，有序消費(fèi)模式返回ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT后，消費(fèi)者會立馬消費(fèi)這條消息，而使用并發(fā)消費(fèi)模式，返回ConsumeConcurrentlyStatus.RECONSUME_LATER后，要過好幾秒甚至十幾秒才會再次消費(fèi)。

我是在只有一條消息的情況下測試的。更重要的區(qū)別是，

返回ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT并不會增加消息的消費(fèi)次數(shù)，mq消息有個默認(rèn)最大消費(fèi)次數(shù)16，消費(fèi)次數(shù)到了以后，這條消息會進(jìn)入死信隊(duì)列，這個最大消費(fèi)次數(shù)是可以在mqadmin中設(shè)置的。

mqadmin updateSubGroup -n 127.0.0.1:9876 -c DefaultCluster -g MonitorCumsumerGroupName -r 3

我測試后發(fā)現(xiàn)，并發(fā)模式下返回ConsumeConcurrentlyStatus.RECONSUME_LATER，同一個消息到達(dá)最大消費(fèi)次數(shù)之后就不會再出現(xiàn)了。這說明有序消費(fèi)模式可能并沒有這個機(jī)制，這意味著你再有序消費(fèi)模式下拋出固定異常，那么這條異常信息將會被永遠(yuǎn)消費(fèi)，并且很可能會影響之后正常的消息。下面依然做個試驗(yàn)

Map<String, Integer> map = new HashMap<>();//保存消息錯誤消費(fèi)次數(shù)
new MessageListenerOrderly() {
 
	@Override
	public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
			ConsumeOrderlyContext context) {
        try {
        	if(1 == 1)
        			throw new Exception();
            return ConsumeOrderlyStatus.SUCCESS;  
        }catch (Exception e) {
        	MessageExt msg = msgs.get(0);
			if(map.containsKey(msg.getKeys())) {//消息每消費(fèi)一次，加1
			    map.put(msg.getKeys(), map.get(msg.getKeys()) + 1);
			}else {
			    map.put(msg.getKeys(), 1);
			}
			logger.info(msg.getKeys()+":"+map.get(msg.getKeys()));
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
	}	
}

發(fā)送了十條消息