建站
咨詢
售后
建站咨詢
新聞中心
這篇文章給大家介紹如何理解RocketMQ消費(fèi)位置，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。
創(chuàng)新互聯(lián)公司專注于臺(tái)江網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗(yàn)。 熱誠(chéng)為您提供臺(tái)江營(yíng)銷型網(wǎng)站建設(shè)，臺(tái)江網(wǎng)站制作、臺(tái)江網(wǎng)頁(yè)設(shè)計(jì)、臺(tái)江網(wǎng)站官網(wǎng)定制、微信小程序服務(wù)，打造臺(tái)江網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供臺(tái)江網(wǎng)站排名全網(wǎng)營(yíng)銷落地服務(wù)。
RocketMQ創(chuàng)建消費(fèi)者的時(shí)指定了Topic主題及Tag,我們發(fā)現(xiàn)新創(chuàng)建的消費(fèi)者消費(fèi)不了歷史的數(shù)據(jù)，只能消費(fèi)掉創(chuàng)建以后消費(fèi)者發(fā)送的數(shù)據(jù)。這是什么原因，我們能把所有的消息都消費(fèi)嗎？，我們可以指定需要消費(fèi)的消息的時(shí)間嗎？答案是肯定的，下面我們具體分析一下。
前提：我們討論是集群模式下的，廣播模式也是一樣的，只是示例代碼我們用集群模式來(lái)討論。
消息消費(fèi)的位置目前提供了三種方式CONSUME_FROM_LAST_OFFSET（隊(duì)列尾部消費(fèi)）、CONSUME_FROM_FIRST_OFFSET（隊(duì)列頭部消費(fèi)）、CONSUME_FROM_TIMESTAMP（指定消費(fèi)時(shí)間點(diǎn)）。
public enum ConsumeFromWhere {
CONSUME_FROM_LAST_OFFSET,
@Deprecated
CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST,
@Deprecated
CONSUME_FROM_MIN_OFFSET,
@Deprecated
CONSUME_FROM_MAX_OFFSET,
CONSUME_FROM_FIRST_OFFSET,
CONSUME_FROM_TIMESTAMP,
}分析源碼我們看到有6種方式，其他三種已經(jīng)廢棄掉了，不做討論。
1、從隊(duì)列尾部消費(fèi)（默認(rèn)）
我們從新創(chuàng)建一個(gè)消費(fèi)組來(lái)消費(fèi)某個(gè)主題下的消息時(shí)，歷史消息沒(méi)有被消費(fèi)，當(dāng)生產(chǎn)者重新發(fā)送消息時(shí)則會(huì)接收到最新的，我們分析下其在哪設(shè)置的。
當(dāng)創(chuàng)建消費(fèi)者的時(shí)候內(nèi)置了一些參數(shù)，從隊(duì)列尾部消費(fèi)。
從隊(duì)列尾部消費(fèi)導(dǎo)致歷史消息消費(fèi)不了，會(huì)丟失一部分?jǐn)?shù)據(jù)，如果僅僅是狀態(tài)數(shù)據(jù)則可以這樣設(shè)置，如果是業(yè)務(wù)數(shù)據(jù)導(dǎo)致數(shù)據(jù)丟失。
對(duì)于設(shè)置這個(gè)參數(shù)僅對(duì)于消費(fèi)組第一次創(chuàng)建時(shí)生效，后面再次設(shè)置不生效，因?yàn)樵撓M(fèi)組在服務(wù)端已經(jīng)記錄了消費(fèi)的進(jìn)度，已有進(jìn)度位置。
查看消費(fèi)進(jìn)度文件的位置,我們根據(jù)上幾節(jié)的內(nèi)容查看TopicTest主題下的這個(gè)consumer_test_clustering消費(fèi)組的消息消費(fèi)的進(jìn)度。查看Broker-a服務(wù)器節(jié)點(diǎn)上的信息。
查看消費(fèi)的消費(fèi)進(jìn)度先根據(jù)可視化界面查看
查看服務(wù)器文件上的消費(fèi)進(jìn)度信息：/usr/local/rocketmq-all-4.2.0/store/config/consumerOffset.json
2、從隊(duì)列頭部消費(fèi)
編寫(xiě)Consumer
public class Consumer1 {
public static void main(String[] args){
try {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setConsumerGroup("consumer_first_offset");
consumer.setNamesrvAddr("10.10.12.203:9876;10.10.12.204:9876");
consumer.subscribe("TopicTest", "*");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerConcurrently(){
@Override
public ConsumeConcurrentlyStatus consumeMessage(List paramList,
ConsumeConcurrentlyContext paramConsumeConcurrentlyContext) {
try {
for(MessageExt msg : paramList){
String msgbody = new String(msg.getBody(), "utf-8");
SimpleDateFormat sd = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss");
Date date = new Date(msg.getStoreTimestamp());
System.out.println("Consumer1=== 存入時(shí)間 : "+ sd.format(date) +" == MessageBody: "+ msgbody);//輸出消息內(nèi)容
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再試
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消費(fèi)成功
}
});
consumer.start();
System.out.println("Consumer1===啟動(dòng)成功!");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
} 設(shè)置了消費(fèi)組：consumer.setConsumerGroup("consumer_first_offset");
設(shè)置了消費(fèi)位置：consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
查看其結(jié)果
從頭開(kāi)始消費(fèi)是指目前還儲(chǔ)存在broker的上的消息全部消費(fèi)一遍，因?yàn)镽ocketMQ會(huì)將消息持久化到磁盤(pán)文件中，時(shí)間長(zhǎng)就會(huì)導(dǎo)致磁盤(pán)文件會(huì)很多，RocketMQ有一種機(jī)制，只是保留一段時(shí)間的消息，之前的消息會(huì)刪除，可以指定時(shí)間點(diǎn)刪除（無(wú)論消息是否被消費(fèi)，到時(shí)間點(diǎn)文件都會(huì)被刪除）
3、從指定時(shí)間點(diǎn)消費(fèi)
消費(fèi)者代碼
public class Consumer1 {
public static void main(String[] args){
try {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
consumer.setConsumerGroup("consumer_time_offset");
consumer.setNamesrvAddr("10.10.12.203:9876;10.10.12.204:9876");
consumer.subscribe("TopicTest", "*");
//可以設(shè)置從什么時(shí)間開(kāi)始消費(fèi)，配合setConsumeTimestamp一起使用默認(rèn)半小時(shí)之前的,格式y(tǒng)yyyMMddhhmmss
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - 1800000L));
consumer.registerMessageListener(new MessageListenerConcurrently(){
@Override
public ConsumeConcurrentlyStatus consumeMessage(List paramList,
ConsumeConcurrentlyContext paramConsumeConcurrentlyContext) {
try {
for(MessageExt msg : paramList){
String msgbody = new String(msg.getBody(), "utf-8");
SimpleDateFormat sd = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss");
Date date = new Date(msg.getStoreTimestamp());
System.out.println("Consumer1=== 存入時(shí)間 : "+ sd.format(date) +" == MessageBody: "+ msgbody);//輸出消息內(nèi)容
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再試
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消費(fèi)成功
}
});
consumer.start();
System.out.println("Consumer1===啟動(dòng)成功!");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
} 設(shè)置消費(fèi)位置：consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP); 設(shè)置消費(fèi)的時(shí)間點(diǎn)：consumer.setConsumeTimestamp("20181222171201");
如果從消息進(jìn)度服務(wù)OffsetStore讀取到MessageQueue中的偏移量大于等于0，則使用讀取到的偏移量，只有讀取到的偏移量小于0時(shí)上述策略才會(huì)生效。
關(guān)于如何理解RocketMQ消費(fèi)位置就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。
文章標(biāo)題：如何理解RocketMQ消費(fèi)位置
網(wǎng)址分享：http://www.dlmjj.cn/article/jpdpjc.html
