RocketMQ创建消费者的时指定了Topic主题及Tag,我们发现新创建的消费者消费不了历史的数据,只能消费掉创建以后消费者发送的数据。这是什么原因,我们能把所有的消息都消费吗?,我们可以指定需要消费的消息的时间吗?答案是肯定的,下面我们具体分析一下。
前提:我们讨论是集群模式下的,广播模式也是一样的,只是示例代码我们用集群模式来讨论。
消息消费的位置目前提供了三种方式CONSUME_FROM_LAST_OFFSET(队列尾部消费)、CONSUME_FROM_FIRST_OFFSET(队列头部消费)、CONSUME_FROM_TIMESTAMP(指定消费时间点)。
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种方式,其他三种已经废弃掉了,不做讨论。
1、从队列尾部消费(默认)
我们从新创建一个消费组来消费某个主题下的消息时,历史消息没有被消费,当生产者重新发送消息时则会接收到最新的,我们分析下其在哪设置的。
当创建消费者的时候内置了一些参数,从队列尾部消费。
从队列尾部消费导致历史消息消费不了,会丢失一部分数据,如果仅仅是状态数据则可以这样设置,如果是业务数据导致数据丢失。
对于设置这个参数仅对于消费组第一次创建时生效,后面再次设置不生效,因为该消费组在服务端已经记录了消费的进度,已有进度位置。
查看消费进度文件的位置,我们根据上几节的内容查看TopicTest主题下的这个consumer_test_clustering消费组的消息消费的进度。查看Broker-a服务器节点上的信息。
查看消费的消费进度先根据可视化界面查看
查看服务器文件上的消费进度信息:/usr/local/rocketmq-all-4.2.0/store/config/consumerOffset.json
2、从队列头部消费
编写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<MessageExt> 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=== 存入时间 : "+ sd.format(date) +" == MessageBody: "+ msgbody);//输出消息内容
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
}
});
consumer.start();
System.out.println("Consumer1===启动成功!");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
设置了消费组:
consumer.setConsumerGroup(“consumer_first_offset”);
设置了消费位置:
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
查看其结果
从头开始消费是指目前还储存在broker的上的消息全部消费一遍,因为RocketMQ会将消息持久化到磁盘文件中,时间长就会导致磁盘文件会很多,RocketMQ有一种机制,只是保留一段时间的消息,之前的消息会删除,可以指定时间点删除(无论消息是否被消费,到时间点文件都会被删除)
3、从指定时间点消费
消费者代码
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", "*");
//可以设置从什么时间开始消费,配合setConsumeTimestamp一起使用默认半小时之前的,格式yyyyMMddhhmmss
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - 1800000L));
consumer.registerMessageListener(new MessageListenerConcurrently(){
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> 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=== 存入时间 : "+ sd.format(date) +" == MessageBody: "+ msgbody);//输出消息内容
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
}
});
consumer.start();
System.out.println("Consumer1===启动成功!");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
设置消费位置:consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP); 设置消费的时间点:consumer.setConsumeTimestamp(“20181222171201”);
如果从消息进度服务OffsetStore读取到MessageQueue中的偏移量大于等于0,则使用读取到的偏移量,只有读取到的偏移量小于0时上述策略才会生效。
转载于:https://my.oschina.net/mingxungu/blog/3083953