前言

首先来明确一下 Offset 的含义， RocketMQ 中， 一 种类型的消息会放到 一 个 Topic 里，为了能够并行， 一般一个 Topic 会有多个 Message Queue (也可以 设置成一个)， Offset是指某个 Topic下的一条消息在某个 Message Queue里的 位置，通过 Offset的值可以定位到这条消息，或者指示 Consumer从这条消息开始向后继续处理 。

Offset主要分为本地文件类型和 Broker代存 的类型两种 。

Rocketmq集群有两种消费模式

  • 默认是 CLUSTERING（集群） 模 式，也就是同一个 Consumer group 里的多个消费者每人消费一部分，各自收到 的消息内容不一样 。 这种情况下，由 Broker 端存储和控制 Offset 的值，使用 RemoteBrokerOffsetStore 结构 。
  • BROADCASTING（广播）模式下，每个 Consumer 都收到这个 Topic 的全部消息，各个 Consumer 间相互没有干扰， RocketMQ 使 用 LocalfileOffsetStore，把 Offset存到本地 。

概念

  • message queue 是无限长的数组，一条消息进来下标就会涨1，下标就是 offset，消息在某个 MessageQueue 里的位置，通过 offset 的值可以定位到这条消息，或者指示 Consumer 从这条消息开始向后处理。
  • message queue 中的 maxOffset 表示消息的最大 offset，maxOffset 并不是最新的那条消息的 offset，而是最新消息的 offset+1，minOffset 则是现存在的最小 offset。
  • fileReserveTime=48 默认消息存储48小时后，消费会被物理地从磁盘删除，message queue 的 minOffset 也就对应增长。所以比 minOffset 还要小的那些消息已经不在 broker上了，就无法被消费

类型(父类是OffsetStore)：

  • 本地文件类型
    • DefaultMQPushConsumer 的 BROADCASTING 广播模式，各个 Consumer 没有互相干扰，使用 LocalFileOffsetStore，把 Offset 存储在本地
  • Broker 代存储类型
    • DefaultMQPushConsumer 的 CLUSTERING 集群模式，由 Broker 端存储和控制 Offset 的值，使用 RemoteBrokerOffsetStore

作用

  • 主要是记录消息的偏移量，有多个消费者进行消费
  • 集群模式下采用 RemoteBrokerOffsetStore，broker 控制 offset 的值
  • 广播模式下采用 LocalFileOffsetStore，消费端存储

建议采用 pushConsumer，RocketMQ 自动维护 OffsetStore，如果用另外一种 pullConsumer 需要自己进行维护 OffsetStore

消息存储 CommitLog

消息存储是由 ConsumeQueue 和 CommitLog 配合完成

  • ConsumeQueue 是逻辑队列，CommitLog 是真正存储消息文件的，ConsumeQueue 存储的是指向物理存储的地址。Topic 下的每个 message queue 都有对应的 ConsumeQueue 文件，内容也会被持久化到磁盘。默认地址：store/consumequeue/{topicName}/{queueid}/fileName
  • CommitLog：存储消息真正内容的文件。
  • 生成规则：
    • 每个文件的默认1G =1024 * 1024 * 1024，commitlog 的文件名 fileName，名字长度为20位，左边补零，剩余为起始偏移量；比如 00000000000000000000 代表了第一个文件，起始偏移量为0，文件大小为1G=1 073 741 824 Byte；当这个文件满了，第二个文件名字为00000000001073741824，起始偏移量为1073741824，消息存储的时候会顺序写入文件，当文件满了则写入下一个文件。
    • 判断消息存储在哪个 CommitLog 上
      • 例如 1073742827 为物理偏移量，则其对应的相对偏移量为 1003 = 1073742827 - 1073741824，并且该偏移量位于第二个 CommitLog。

Broker 里面一个 Topic 里面有多个 MesssageQueue，每个 MessageQueue 对应一个 ConsumeQueue，ConsumeQueue 里面记录的是消息在 CommitLog 里面的物理存储地址。

IndexFile 消息索引文件

    ConsumerQueue是通过偏移量offset去CommitLog文件中查找消息，但实际工作应用中，我们想查找某条具体的消息，并不知道offset值，那该怎么办呢？那IndexFile作用就来了。（如果我们需要根据消息ID，来查找消息，consumequeue 中没有存储消息ID,如果不采取其他措施，又得遍历 commitlog文件了，indexFile就是为了解决这个问题的文件）。
    IndexFile是消息索引文件，如果一个生产者发送的消息包含key值的话，会使用IndexFile存储消息索引，主要用于使用key来查询消息。文件的内容结构如图

在Broker端，通过Key来计算Hash槽的位置，从而找到Index索引数据。从Index索引中拿到消息的物理偏移量，最后根据这个物理偏移量，直接到CommitLog文件中去找就可以了。另外说明下，通过IndexFile来查找消息的方法不影响RocketMQ的正常生产-消费流程，它只是查询定位消息的方法而已。

offset

在rocketMQ中，offset用来管理每个消费队列的不同消费组的消费进度。对offset的管理分为本地模式和远程模式，本地模式是以文本文件的形式存储在客户端，而远程模式是将数据保存到broker端，对应的数据结构分别为LocalFileOffsetStore和RemoteBrokerOffsetStore。
默认情况下，当消费模式为广播模式时，offset使用本地模式存储，因为每条消息会被所有的消费者消费，每个消费者管理自己的消费进度，各个消费者之间不存在消费进度的交集；当消费模式为集群消费时，则使用远程模式管理offset，消息会被多个消费者消费，不同的是每个消费者只负责消费其中部分消费队列，添加或删除消费者，都会使负载发生变动，容易造成消费进度冲突，因此需要集中管理。同时，RocketMQ也提供接口供用户自己实现offset管理（实现OffsetStore接口）。
生产环境上一般使用集群模式，本文主要记录集群模式下offset的管理，即RemoteBrokerOffsetStore。

broke端

offset的存储与加载

rocketMQ的broker端中，offset的是以json的形式持久化到磁盘文件中，文件路径为${user.home}/store/config/consumerOffset.json。其内容示例如下：

{
    "offsetTable": {
        "test-topic@test-group": {
            "0": 88526, 
            "1": 88528
        }
    }
}

broker端启动后，会调用BrokerController.initialize()方法，方法中会对offset进行加载，consumerOffsetManager.load()。获取文件内容后，序列化为ConsumerOffsetManager对象，实质是其属性ConcurrentMap<String，ConcurrentMap<Integer, Long>> offsetTable，offsetTable的数据结构为ConcurrentMap，是一个线程安全的容器，key的形式为topic@group（每个topic下不同消费组的消费进度），value也是一个ConcurrentMap，key为queueId，value为消费位移（这里不是offset而是位移）。通过对全局ConsumerOffsetManager对象就可以对各个topic下不同消费组的消费位移进行获取与管理。

/**ConsumerOffsetManager.offsetTable*/
private ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> offsetTable =
        new ConcurrentHashMap<String, ConcurrentMap<Integer, Long>>(512);

/**ConsumerOffsetManager.decode*/
public void decode(String jsonString) {
        if (jsonString != null) {
            // 序列化成功后复制给全局ConsumerOffsetManager对象
            ConsumerOffsetManager obj = RemotingSerializable.fromJson(jsonString, ConsumerOffsetManager.class);
            if (obj != null) {
                this.offsetTable = obj.offsetTable;
            }
        }
    }

commitLog与offset

如下图所示，producer发送消息到broker之后，会将消息具体内容持久化到commitLog文件中，再分发到topic下的消费队列consume Queue，消费者提交消费请求时，broker从该consumer负责的消费队列中根据请求参数起始offset获取待消费的消息索引信息，再从commitLog中获取具体的消息内容返回给consumer。在这个过程中，consumer提交的offset为本次请求的起始消费位置，即beginOffset；consume Queue中的offset定位了commitLog中具体消息的位置。
consume Queue中每个消息索引信息长度为20bytes，包括8位长度的offset，记录commitLog中消息内容的位移；4位长度的size，记录具体消息内容的长度；8位长度的tagHashCode，记录消息的tag的哈希值（订阅时如果指定tag，会根据HashCode快速查找订阅的消息）

nextBeginOffset

对于consumer的消费请求处理（PullMessageProcessor.processRequest()），除了待消费的消息内容，broker在responseHeader（PullMessageResponseHeader）附带上当前消费队列的最小offset（minOffset）、最大offset（maxOffset）、及下次拉取的起始offset（nextBeginOffset）。

  • minOffset、maxOffset是当前消费队列consumeQueue记录的最小及最大的offset信息。
  • nextBeginOffset是consumer下次拉取消息的offset信息，即consumer对该consumeQueue的消费进度。

其中nextBeginOffset是consumer在下一轮消息拉取时offset的重要依据，无论当次拉取的消息消费是否正常，nextBeginOffset都不会回滚，这是因为rocketMQ对消费异常的消息的处理是将消息重新发回broker端的重试队列（会为每个topic创建一个重试队列，以%RERTY%开头），达到重试时间后将消息投递到重试队列中进行消费重试。对消费异常的处理不是通过offset回滚，这使得客户端简化了offset的管理。

client端

offset初始化

consumer启动过程中（Consumer主函数默认调用DefaultMQPushConsumer.start()方法）根据MessageModel选择对应的offsetStore，然后调用offsetStore.load()对offset进行加载，LocalFileOffsetStore是对本地文件的加载，而RemotebrokerOffsetStore是没有本地文件的，因此load()方法没有实现。在rebalance完成对messageQueue的分配之后会对messageQueue对应的消费位置offset进行更新。

/** RebalanceImpl */
/**
doRebalance() -> rebalanceByTopic() -> updateProcessQueueTableInRebalance() 
-> computePullFromWhere()
*/
private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
        final boolean isOrder) {
    // (省略部分代码)负载均衡获取当前consumer负责的消息队列后对processQueue进行筛选，删除processQueue不必要的messageQueue
    
    // 获取topic下consumer消息拉取列表，List<PullRequest>
    List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
    for (MessageQueue mq : mqSet) {
        if (!this.processQueueTable.containsKey(mq)) {
                if (isOrder && !this.lock(mq)) {
                    log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
                    continue;
                }

                // 删除messageQueue旧的offset信息
                this.removeDirtyOffset(mq);
                ProcessQueue pq = new ProcessQueue();
                // 获取nextOffset，即更新当前messageQueue对应请求的offset
                long nextOffset = this.computePullFromWhere(mq);
                if (nextOffset >= 0) {
                    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                    if (pre != null) {
                        log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
                    } else {
                        log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                        PullRequest pullRequest = new PullRequest();
                        pullRequest.setConsumerGroup(consumerGroup);
                        pullRequest.setNextOffset(nextOffset);
                        pullRequest.setMessageQueue(mq);
                        pullRequest.setProcessQueue(pq);
                        pullRequestList.add(pullRequest);
                        changed = true;
                    }
                } else {
                    log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
                }
            }
    }
    
}

Push模式下，computePullFromWhere()方法的实现类为RebalancePushImpl.class。根据配置信息consumeFromWhere进行不同的操作。ConsumeFromWhere的类型枚举如下，其中有三个已经被标记为Deprecated（基于rocketmq-all 4.6.0版本）

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,
}
  • CONSUME_FROM_LAST_OFFSET

从最新的offset开始消费。
获取consumer对当前消息队列messageQueue的消费进度lastOffset，如果lastOffset>=0，从lastOffset开始消费；如果lastOffset小于0说明是first start，没有offset信息，topic为重试topic时从0开始消费，否则请求获取该消息队列对应的消费队列consumeQueue的最大offset（maxOffset），从maxOffset开始消费

  • CONSUME_FROM_FIRST_OFFSET

从第一个offset开始消费。
获取consumer对当前消息队列messageQueue的消费进度lastOffset，如果lastOffset>=0，从lastOffset开始消费；
否则从0开始消费。

  • CONSUME_FROM_TIMESTAMP

获取consumer对当前消息队列messageQueue的消费进度lastOffset，如果lastOffset>=0，从lastOffset开始消费；
当lastOffset<0，如果为重试topic，获取consumeQueue的最大offset；否则获取ConsumeTimestamp（consumer启动时间），根据时间戳请求查找offset。

上述三种消费位置的设置流程有一个共同点，都请求获取consumer对当前消息队列messageQueue的消费进度lastOffset，如果lastOffset不小于0，则从lastOffset开始消费。这也是有时候设置了CONSUME_FROM_FIRST_OFFSET却不是从0开始重新消费的原因，rocketMQ减少了由于配置原因造成的重复消费。

对于lastOffset、maxOffset、时间戳查找offset都是通过MQClientAPIImpl提供的接口进行查询的，MQClientAPIImplclient对broker请求的封装类，使用Netty进行异步请求，对应的RequestCode分别为RequestCode.QUERY_CONSUMER_OFFSET、RequestCode.GET_MAX_OFFSET、RequestCode.SEARCH_OFFSET_BY_TIMESTAMP。

/** RebalancePushImpl */
public long computePullFromWhere(MessageQueue mq) {
        long result = -1;
        final ConsumeFromWhere consumeFromWhere = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere();
        final OffsetStore offsetStore = this.defaultMQPushConsumerImpl.getOffsetStore();
        switch (consumeFromWhere) {
            case CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST:
            case CONSUME_FROM_MIN_OFFSET:
            case CONSUME_FROM_MAX_OFFSET:
            case CONSUME_FROM_LAST_OFFSET: {
                // 从broker获取当前消费队列offset
                long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                if (lastOffset >= 0) {
                    result = lastOffset;
                }
                // First start,no offset
                else if (-1 == lastOffset) {
                    if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        result = 0L;
                    } else {
                        try {
                            // 获取消费队列最大offset
                            result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
                        } catch (MQClientException e) {
                            result = -1;
                        }
                    }
                } else {
                    result = -1;
                }
                break;
            }
            case CONSUME_FROM_FIRST_OFFSET: {
                // 先查询当前消费队列消费进度
                long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                if (lastOffset >= 0) {
                    result = lastOffset;
                }
                // 当前消费队列消费进度小于0，则从0开始
                else if (-1 == lastOffset) {
                    result = 0L;
                } else {
                    result = -1;
                }
                break;
            }
            case CONSUME_FROM_TIMESTAMP: {
                // 同样也是先查询当前消费队列消费进度
                long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                if (lastOffset >= 0) {
                    result = lastOffset;
                } else if (-1 == lastOffset) {
                    if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        try {
                            result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
                        } catch (MQClientException e) {
                            result = -1;
                        }
                    } else {
                        try {
                            // 获取consumer启动时间
                            long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(),
                                UtilAll.YYYYMMDDHHMMSS).getTime();
                            // 根据时间戳获取offset信息
                            result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
                        } catch (MQClientException e) {
                            result = -1;
                        }
                    }
                } else {
                    result = -1;
                }
                break;
            }

            default:
                break;
        }

        return result;
    }

offset提交更新

consumer从broker拉取消息后，会将消息的扩展信息MessageExt存放到ProcessQueue的属性TreeMap<Long, MessageExt> msgTreeMap中，key值为消息对应的queueOffset，value为扩展信息（包括queueID等）。并发消费模式下（Concurrently），获取的待消费消息会分批提交给消费线程进行消费，默认批次为1，即每个消费线程消费一条消息。消费完成后调用ConsumerMessageConcurrentlyService.processConsumeResult()方法对结果进行处理：消费成功确认ack，消费失败发回broker进行重试。之后便是对offset的更新操作。
首先是调用ProcessQueue.removeMessage()方法，将已经消费完成的消息从msgTreeMap中根据queueOffset移除，然后判断当前msgTreeMap是否为空，不为空则返回当前msgTreeMap第一个元素，即offset最小的元素，否则返回-1。
如果removeMessage()返回的offset大于0，则更新到offsetTable中。offsetTable的结构为ConcurrentMap<MessageQueue, AtomicLong> offsetTable，是一个线程安全的Map，key为MessageQueue，value为AtomicLong对象，值为offset，记录当前messageQueue的消费位移。

/** ConsumeMessageConcurrentlyService.class */
public void processConsumeResult(
        final ConsumeConcurrentlyStatus status,final ConsumeConcurrentlyContext context,final ConsumeRequest consumeRequest) {
    // .... （省略部分代码）根据消费结果判断是否需要发回broker重试
    
    // 在msgTreeMap中删除msg，标记当前消息已被消费，msgTreeMap不为空返回当前msgTreeMap中最小的offset
    long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
    
    // 更新offsetTable中的消费位移，offsetTable记录每个messageQueue的消费进度
    // updateOffset()的最后一个参数increaseOnly为true，表示单调增加，新值要大于旧值
    if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
        this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
     }
}

/** ProcessQueue.class */
public long removeMessage(final List<MessageExt> msgs) {
        long result = -1;
        final long now = System.currentTimeMillis();
        try {
            this.lockTreeMap.writeLock().lockInterruptibly();
            this.lastConsumeTimestamp = now;
            try {
                if (!msgTreeMap.isEmpty()) {
                    result = this.queueOffsetMax + 1;
                    int removedCnt = 0;
                    // // 从msgTreeMap中删除该批次的msg
                    for (MessageExt msg : msgs) {
                        MessageExt prev = msgTreeMap.remove(msg.getQueueOffset());
                        if (prev != null) {
                            removedCnt--;
                            msgSize.addAndGet(0 - msg.getBody().length);
                        }
                    }
                    msgCount.addAndGet(removedCnt);

                    // 删除后当前msgTreeMap不为空，返回第一个元素，即最小的offset
                    if (!msgTreeMap.isEmpty()) {
                        result = msgTreeMap.firstKey();
                    }
                }
            } finally {
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (Throwable t) {
            log.error("removeMessage exception", t);
        }

        return result;
    }

/** RemoteBrokerOffsetStore */
public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
        if (mq != null) {
            AtomicLong offsetOld = this.offsetTable.get(mq);
            if (null == offsetOld) {
                // offsetTable中不存在mq对应的记录
                // putIfAbsent 如果传入key对应的value已存在，则返回存在的value，不替换；如果不存在，则新增，返回null
                offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));
            }

            // offsetTable存在记录，替换，这里increaseOnly为true，offsetOld<offset才替换
            if (null != offsetOld) {
                if (increaseOnly) {
                    MixAll.compareAndIncreaseOnly(offsetOld, offset);
                } else {
                    offsetOld.set(offset);
                }
            }
        }
    }

到这里一条消息的消费流程已经结束，offset更新到了本地缓存offsetTable，而将offset上传到broker是由定时任务执行的。MQClientInstance.start()会启动客户端相关的定时任务，包括NameService通信、offset提交等。

/** MQClientInstance.startScheduledTask() */
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    // 提交offset至broker
                    MQClientInstance.this.persistAllConsumerOffset();
                } catch (Exception e) {
                    log.error("ScheduledTask persistAllConsumerOffset exception", e);
                }
            }
        }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

LocalFileOffsetStore模式下，将offset信息转化成json保存到本地文件中；RemoteBrokerOffsetStore则offsetTable将需要提交的MessageQueue的offset信息通过MQClientAPIImpl提供的接口updateConsumerOffsetOneway()提交到broker进行持久化存储。
另一种情况，当应用正常关闭时，consumer的shutdown()方法会主动触发一次持久化offset到broker的操作。

client对offset的更新是在消息消费完成后将offset更新到offsetTable，再由定时任务进行持久化。这个过程有需要注意的地方。

  • 由于是先消费再更新offset，因此存在消费完成后更新offset失败，但这种情况出现的概率比较低，更新offset只是写到缓存中，是一个简单的内存操作，出错的可能性较低。
  • 由于offset先存到内存中，再由定时任务每隔10s提交一次，存在丢失的风险，比如当前client宕机等，从而导致更新后的offset没有提交到broker，再次负载时会重复消费。因此consumer的消费业务逻辑需要保证幂等性。

并发消费时offset的更新

问题：consumer从broker拉取的待消费消息时批量的（默认情况下pullBatchSize=32），并发消费时，offset的更新不是按大小顺序的，比如拉取消息m1到m10，m1可能是最后消费完成的，那提交的offset的正确性如何保证？m10 offset的更新不会导致m1会误认为已消费完成。
上一小节提到消费完成后，会将线程消费的批次消息从msgTreeMap中删除，并返回当前msgTreeMap的第一个元素，也就是拉取批次最小的offset，offsetTable更新的offset一直会是拉取批次中未消费的最小的offset值。也就是m1未消费完成，m10消费完成的情况下，更新到offsetTable的当前messageQueue的消费进度为m1对应的offset值。

因此，offsetTable中存放的可能不是messageQueue真正消费的offset的最大值，但是consumer拉取消息时使用的是上一次拉取请求返回的nextBeginOffset，并不是依据offsetTable，正常情况下不会重复拉取数据。当发生宕机等异常时，与offsetTable未提交宕机异常一样，需要通过业务流程来保证幂等性。业务流程的幂等性是rocketMQ一直强调的。

注意

nextBeginOffset、消费进度offset两者和拉取消息的关系，正常情况下消费者按nextBeginOffset一批一批的拉取消息，当异常出现，即offsetTable中存放的消费进度offset卡在某个最小值，而消费者仍在按nextBeginOffset一批一批的拉取消息，直到消息触发RocketMQ 消费端限流机制才停止消息拉取，而此时生产者仍在继续工作，进而造成消息积压。（这里有个疑问，官网讲消费位点的保存和恢复是基于 Apache RocketMQ 服务端的存储实现，和任何消费者无关。因此 Apache RocketMQ 支持跨消费者的消费进度恢复。那么它和消费者上报的offset 和消费位点 是什么关系呢？？？？？？？？？？？？？？）

以下是官网消费进度管理知识点

消费进度管理

Apache RocketMQ 通过消费位点管理消费进度，本文为您介绍 Apache RocketMQ 的消费进度管理机制。

背景信息​

Apache RocketMQ 的生产者和消费者在进行消息收发时，必然会涉及以下场景，消息先生产后订阅或先订阅后生产。这两种场景下，消费者客户端启动后从哪里开始消费？如何标记已消费的消息？这些都是由 Apache RocketMQ 的消费进度管理机制来定义的。

通过了解 Apache RocketMQ 的消费进度管理机制，可以帮助您解答以下问题：

  • 消费者启动后从哪里开始消费消息？

  • 消费者每次消费成功后如何标记消息状态，确保下次不会再重复处理该消息？

  • 某消息被指定消费者消费过一次后，如果业务出现异常需要做故障恢复，该消息能否被重新消费？

消费进度原理​

消息位点（Offset）

参考 Apache RocketMQ 主题和队列的定义，消息是按到达服务端的先后顺序存储在指定主题的多个队列中，每条消息在队列中都有一个唯一的Long类型坐标，这个坐标被定义为消息位点。

任意一个消息队列在逻辑上都是无限存储，即消息位点会从0到Long.MAX无限增加。通过主题、队列和位点就可以定位任意一条消息的位置，具体关系如下图所示：

消息位点

Apache RocketMQ 定义队列中最早一条消息的位点为最小消息位点（MinOffset）；最新一条消息的位点为最大消息位点（MaxOffset）。虽然消息队列逻辑上是无限存储，但由于服务端物理节点的存储空间有限， Apache RocketMQ 会滚动删除队列中存储最早的消息。因此，消息的最小消费位点和最大消费位点会一直递增变化。

消费位点更新

消费位点（ConsumerOffset）

Apache RocketMQ 领域模型为发布订阅模式，每个主题的队列都可以被多个消费者分组订阅。若某条消息被某个消费者消费后直接被删除，则其他订阅了该主题的消费者将无法消费该消息。

因此，Apache RocketMQ 通过消费位点管理消息的消费进度。每条消息被某个消费者消费完成后不会立即在队列中删除，Apache RocketMQ 会基于每个消费者分组维护一份消费记录，该记录指定消费者分组消费某一个队列时，消费过的最新一条消息的位点，即消费位点。

当消费者客户端离线，又再次重新上线时，会严格按照服务端保存的消费进度继续处理消息。如果服务端保存的历史位点信息已过期被删除，此时消费位点向前移动至服务端存储的最小位点。

信息

消费位点的保存和恢复是基于 Apache RocketMQ 服务端的存储实现，和任何消费者无关。因此 Apache RocketMQ 支持跨消费者的消费进度恢复。

队列中消息位点MinOffset、MaxOffset和每个消费者分组的消费位点ConsumerOffset的关系如下：

消费进度

  • ConsumerOffset≤MaxOffset：

    • 当消费速度和生产速度一致，且全部消息都处理完成时，最大消息位点和消费位点相同，即ConsumerOffset=MaxOffset。

    • 当消费速度较慢小于生产速度时，队列中会有部分消息未消费，此时消费位点小于最大消息位点，即ConsumerOffset<MaxOffset，两者之差就是该队列中堆积的消息量。

  • ConsumerOffset≥MinOffset：正常情况下有效的消费位点ConsumerOffset必然大于等于最小消息位点MinOffset。消费位点小于最小消息位点时是无效的，相当于消费者要消费的消息已经从队列中删除了，是无法消费到的，此时服务端会将消费位点强制纠正到合法的消息位点。

消费位点初始值

消费位点初始值指的是消费者分组首次启动消费者消费消息时，服务端保存的消费位点的初始值。

Apache RocketMQ 定义消费位点的初始值为消费者首次获取消息时，该时刻队列中的最大消息位点。相当于消费者将从队列中最新的消息开始消费。

重置消费位点​

若消费者分组的初始消费位点或当前消费位点不符合您的业务预期，您可以通过重置消费位点调整您的消费进度。

适用场景

  • 初始消费位点不符合需求：因初始消费位点为当前队列的最大消息位点，即客户端会直接从最新消息开始消费。若业务上线时需要消费部分历史消息，您可以通过重置消费位点功能消费到指定时刻前的消息。

  • 消费堆积快速清理：当下游消费系统性能不足或消费速度小于生产速度时，会产生大量堆积消息。若这部分堆积消息可以丢弃，您可以通过重置消费位点快速将消费位点更新到指定位置，绕过这部分堆积的消息，减少下游处理压力。

  • 业务回溯，纠正处理：由于业务消费逻辑出现异常，消息被错误处理。若您希望重新消费这些已被处理的消息，可以通过重置消费位点快速将消费位点更新到历史指定位置，实现消费回溯。

重置功能

Apache RocketMQ 的重置消费位点提供以下能力：

  • 重置到队列中的指定位点。

  • 重置到某一时刻对应的消费位点，匹配位点时，服务端会根据自动匹配到该时刻最接近的消费位点。

使用限制

  • 重置消费位点后消费者将直接从重置后的位点开始消费，对于回溯重置类场景，重置后的历史消息大多属于存储冷数据，可能会造成系统压力上升，一般称为冷读现象。因此，需要谨慎评估重置消费位点后的影响。建议严格控制重置消费位点接口的调用权限，避免无意义、高频次的消费位点重置。

  • Apache RocketMQ 重置消费位点功能只能重置对消费者可见的消息，不能重置定时中、重试等待中的消息。更多信息，请参见定时/延时消息和消费重试。

版本兼容性​

关于消费者分组的消费位点初始值，不同的服务端版本中定义如下：

  • 服务端历史版本（4.x/3.x版本）：消息位点初始值受当前队列消息状态的影响。

  • 服务端5.x版本：明确定义消费位点初始值为消费者获取消息时刻队列中的最大消息位点。

因此，若您将服务端版本从历史版本升级到最新的5.x版本时，需要自行对消费者首次启动时的情况做兼容性判断。

使用建议​

严格控制消费位点重置的权限

重置消费位点会给系统带来额外处理压力，可能会影响新消息的读写性能。 因此该操作请在适用场景下谨慎执行，并提前做好合理性和必要性评估。

原文：RocketMQ -- offset管理

官网：消费进度管理

# java-rocketmq # rocketmq # java
Logo
开放原子开发者工作坊

开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！

更多推荐

OpenLoong项目通过技术监督委员会（TOC）评审

OpenLoong项目通过技术监督委员会（TOC）评审

avatar 开放原子开发者工作坊

开发者谈开源：KWDB开源数据库的未来路径与生态构建实践

开发者谈开源：KWDB开源数据库的未来路径与生态构建实践

avatar 开放原子开发者工作坊

开发者谈开源：洞悉协作创新背后的机遇与挑战

近日，在2024开放原子开发者大会暨首届开源技术学术大会开幕式上，开放原子开源基金会与openKylin、EasyAda、KWDB开源项目举行捐赠签约仪式。 一场捐赠签约仪式，让三个开源项目及其背后的开发者们受到瞩目。本次，我们与“龘”（EasyAda）核心维护者王伶卓开启了对话。

avatar 开放原子开发者工作坊