1、消息消费进度概述
首先简要阐述一下消息消费进度:
消费者订阅消息消费队列(MessageQueue), 当生产者将消息负载发送到 MessageQueue 中时,消费订阅者开始消费消息,消息消费过程中,为了避免重复消费,需要一个地方存储消费进度(消费偏移量)。
消息模式主要分为集群模式、广播模式:
- 集群模式:一条消息被集群中任何一个消费者消费。
- 广播模式:每条消息都被每一个消费者消费。
广播模式,既然每条消息要被每一个消费者消费,则消费进度可以与消费者保存在一起,也就是本地保存,但由于集群模式下,一条消息只能被集群内的一个消费者消费,进度不能保存在消费端,只能集中保存在一个地方,比较合适的是在 Broker 端。
2、消息消费进度存储接口
接下来我们先分析一下消息消费进度接口:OffsetStore。
/*** Offset store interface*/
public interface OffsetStore {/*** Load** @throws MQClientException*/void load() throws MQClientException;/*** Update the offset,store it in memory** @param mq* @param offset* @param increaseOnly*/void updateOffset(final MessageQueue mq, final long offset, final boolean increaseOnly);/*** Get offset from local storage** @param mq* @param type* @return The fetched offset*/long readOffset(final MessageQueue mq, final ReadOffsetType type);/*** Persist all offsets,may be in local storage or remote name server** @param mqs*/void persistAll(final Set<MessageQueue> mqs);/*** Persist the offset,may be in local storage or remote name server** @param mq*/void persist(final MessageQueue mq);/*** Remove offset** @param mq*/void removeOffset(MessageQueue mq);/*** @param topic* @return The cloned offset table of given topic*/Map<MessageQueue, Long> cloneOffsetTable(String topic);/*** @param mq* @param offset* @param isOneway*/void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,MQBrokerException, InterruptedException, MQClientException;
}
入口代码:DefaultMQPushConsumerImpl#start()。
根据消息消费模式(集群模式、广播模式)会创建不同的 OffsetStore 对象。
由于上篇文章,谈到广播模式消息,如果返回 CONSUME_LATER,竟然不会重试,而是直接丢弃,为什么呢?由于这个原因,这次破天荒的从广播模式的OffsetStore开始学习。
2.1 LocalFileOffsetStore (广播模式)
消息进度以本地文件方式保存。源码路径:org.apache.rocketmq.client.consumer.store.LocalFileOffsetStore。
2.1.1 核心属性与构造函数
- LOCAL_OFFSET_STORE_DIR
offset 存储根目录,默认为用户主目录,例如 /home/dingw,可以在消费者启动的JVM参数中,通过 – Drocketmq.client.localOffsetStoreDir=路径。 - groupName
消费组名称。 - storePath
具体的消费进度保存文件名(全路径)。 - offsetTable
内存中的 offfset 进度保持,以 MessageQueue 为键,偏移量为值。
继续看一下构造函数:
LocalFileOffsetStore 首先在 DefaultMQPushConsumerImpl#start 方法中创建,并 执行load方法加载消费进度。接下来结束一下几个关键的实现方法。
2.1.2 load()方法
public void load() throws MQClientException {OffsetSerializeWrapper offsetSerializeWrapper = this.readLocalOffset();if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {offsetTable.putAll(offsetSerializeWrapper.getOffsetTable());for (MessageQueue mq : offsetSerializeWrapper.getOffsetTable().keySet()) {AtomicLong offset = offsetSerializeWrapper.getOffsetTable().get(mq);log.info("load consumer's offset, {} {} {}",this.groupName,mq,offset.get());}}
该方法,主要就是读取 offsets.json 或 offsets.json.bak 中的内容,然后将json转换成map。
然后更新或获取消息队列的消费进度,就是从内存(Map)或 store 中获取,接下来看一下初次保存offsets.json文件。
@Overridepublic void persistAll(Set<MessageQueue> mqs) {if (null == mqs || mqs.isEmpty())return;OffsetSerializeWrapper offsetSerializeWrapper = new OffsetSerializeWrapper();for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {if (mqs.contains(entry.getKey())) {AtomicLong offset = entry.getValue();offsetSerializeWrapper.getOffsetTable().put(entry.getKey(), offset);}}String jsonString = offsetSerializeWrapper.toJson(true);if (jsonString != null) {try {MixAll.string2File(jsonString, this.storePath);} catch (IOException e) {log.error("persistAll consumer offset Exception, " + this.storePath, e);}}
保存逻辑很简单,就没必要一一分析,重点看一下,该方法的调用入口:
MQClientInstance#startScheduledTask
保存逻辑很简单,就没必要一一分析,其调用入口为:MQClientInstance#startScheduledTask。
顺藤摸瓜,原来是一个定时任务,默认消费端启动10秒后,每隔5s的频率持久化一次。
广播模式消费进度存储容易,但其实还是不明白为什么RocketMQ广播模式,如果消费失败,则丢弃,因为广播模式有时候也必须确保每个消费者都成功消费,,通常的场景为,通过MQ刷新本地缓存等。