源码分析RocketMQ之消息ACK机制（消费进度）

1、消息消费进度概述

首先简要阐述一下消息消费进度&＃xff1a;

消费者订阅消息消费队列&＃xff08;MessageQueue), 当生产者将消息负载发送到 MessageQueue 中时&＃xff0c;消费订阅者开始消费消息&＃xff0c;消息消费过程中&＃xff0c;为了避免重复消费&＃xff0c;需要一个地方存储消费进度&＃xff08;消费偏移量&＃xff09;。

消息模式主要分为集群模式、广播模式&＃xff1a;

• 集群模式&＃xff1a;一条消息被集群中任何一个消费者消费。
• 广播模式&＃xff1a;每条消息都被每一个消费者消费。

广播模式&＃xff0c;既然每条消息要被每一个消费者消费&＃xff0c;则消费进度可以与消费者保存在一起&＃xff0c;也就是本地保存&＃xff0c;但由于集群模式下&＃xff0c;一条消息只能被集群内的一个消费者消费&＃xff0c;进度不能保存在消费端&＃xff0c;只能集中保存在一个地方&＃xff0c;比较合适的是在 Broker 端。

2、消息消费进度存储接口

接下来我们先分析一下消息消费进度接口&＃xff1a;OffsetStore。

/*** Offset store interface*/
public interface OffsetStore {/*** Load** &＃64;throws MQClientException*/void load() throws MQClientException;/*** Update the offset,store it in memory** &＃64;param mq* &＃64;param offset* &＃64;param increaseOnly*/void updateOffset(final MessageQueue mq, final long offset, final boolean increaseOnly);/*** Get offset from local storage** &＃64;param mq* &＃64;param type* &＃64;return The fetched offset*/long readOffset(final MessageQueue mq, final ReadOffsetType type);/*** Persist all offsets,may be in local storage or remote name server** &＃64;param mqs*/void persistAll(final Set mqs);/*** Persist the offset,may be in local storage or remote name server** &＃64;param mq*/void persist(final MessageQueue mq);/*** Remove offset** &＃64;param mq*/void removeOffset(MessageQueue mq);/*** &＃64;param topic* &＃64;return The cloned offset table of given topic*/Map cloneOffsetTable(String topic);/*** &＃64;param mq* &＃64;param offset* &＃64;param isOneway*/void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,MQBrokerException, InterruptedException, MQClientException;
}


入口代码&＃xff1a;DefaultMQPushConsumerImpl#start()。