RocketMQ—消息偏移量总结

1 客户端逻辑
1.1 概述
偏移量管理主要是指管理每个消息队列的消费进度：集群模式消费下会将消息队列的消费进度保存在Broker端，广播模式消费下消息队列的消费进度保存在消费者本地。
组件分析：RocketMQ定义了一个接口OffsetStore。它的实现类有两个：RemoteBrokerOffsetStore和LocalFileOffsetStore前者主要是集群消费模式下使用，即与broker进行打交道，将消息队列的消费偏移量通过网络传递给Broker；后者主要是广播消费模式下使用，即直接将消费偏移量存储在消费者所在的本地中。入下图所示：

offsetstore保存在消费者内部客户端ConsumerInner的实现类中的，其初始化创建的时机在内部客户端的start()方法中。offsetstore保存在消费者内部客户端ConsumerInner的实现类中的，其初始化创建的时机在内部客户端的start()方法中。

                if (this.defaultMQPullConsumer.getOffsetStore() != null) {
                    this.offsetStore = this.defaultMQPullConsumer.getOffsetStore();
                } else {
                    switch (this.defaultMQPullConsumer.getMessageModel()) {
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
                            break;
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }
                    this.defaultMQPullConsumer.setOffsetStore(this.offsetStore);
                }

下面主要分析RemoteBrokerOffsetStore的逻辑。
主要是两个逻辑，如下图所示

• 将消息偏移量更新到本地内存中管理消息偏移量的组件
• 将内存中保存的消息偏移量发送给Broker，更新Broker端保存的消息偏移量

1.2 更新消息队列的偏移量
并发消息消费服务中ConsumeMessageConcurrentlyService#processConsumeResult()处理消息消费结果的方法中在消息处理完成以后会调用更新消息队列的偏移量

// 获取偏移量存储实现，并调用其更新偏移量方法更新偏移量  
      long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
        if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
            this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
        }

下面是RemoteBrokerOffsetStore的更新逻辑
将已经确认消费了的偏移量存储偏移量管理器中。此处的更新仅仅是更新了保存每个消息队列的偏移量的map中的值，并没有将偏移量上传到broker。

public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
    if (mq != null) {
        // ConcurrentMap<MessageQueue, AtomicLong>
        // 获取消息队列对应的偏移量
        AtomicLong offsetOld = this.offsetTable.get(mq);
        if (null == offsetOld) {
            // 更新table
            offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));
        }

        if (null != offsetOld) {
            // 是否是只增模式
            if (increaseOnly) {
                MixAll.compareAndIncreaseOnly(offsetOld, offset);
            } else {
                offsetOld.set(offset);
            }
        }
    }
}

1.3 向Broker发送消息偏移量
向服务端发送消息偏移量是通过MQClientInstance中启动的一个定时任务来完成的。
1 在其startScheduledTask方法中开启下列定时任务

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            // 对已消费的消息的偏移量进行持久化
            MQClientInstance.this.persistAllConsumerOffset();
        } catch (Exception e) {
            log.error("ScheduledTask persistAllConsumerOffset exception", e);
        }
    }
}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

2 调用MQClientInstance的persisAllConsumerOffset()方法

private void persistAllConsumerOffset() {
    // 获取所有消费者组对应的内部客户端
    Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<String, MQConsumerInner> entry = it.next();
        MQConsumerInner impl = entry.getValue();
        // 调用内部客户端进行持久化
        impl.persistConsumerOffset();
    }
}

3 调用内部消费者客户端的持久化方法

public void persistConsumerOffset() {
    try {
        this.makeSureStateOK();
        Set<MessageQueue> mqs = new HashSet<MessageQueue>();
        // 获取所有的分配的消息队列
        Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet();
        mqs.addAll(allocateMq);
        // 持久化偏移量
        this.offsetStore.persistAll(mqs);
    } catch (Exception e) {
        log.error("group: " + this.defaultMQPushConsumer.getConsumerGroup() + " persistConsumerOffset exception", e);
    }
}

4 调用偏移量管理器的更新

public void persistAll(Set<MessageQueue> mqs) {
    if (null == mqs || mqs.isEmpty())
        return;

    final HashSet<MessageQueue> unusedMQ = new HashSet<MessageQueue>();

    // 遍历保存消息队列偏移量的map
    for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
        MessageQueue mq = entry.getKey();
        AtomicLong offset = entry.getValue();
        if (offset != null) {
            if (mqs.contains(mq)) {
                try {
                    // 更新到
                    this.updateConsumeOffsetToBroker(mq, offset.get());
                    log.info("[persistAll] Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}",
                            this.groupName,
                            this.mQClientFactory.getClientId(),
                            mq,
                            offset.get());
                } catch (Exception e) {
                    log.error("updateConsumeOffsetToBroker exception, " + mq.toString(), e);
                }
            } else {
                unusedMQ.add(mq);
            }
        }
    }

    if (!unusedMQ.isEmpty()) {
        for (MessageQueue mq : unusedMQ) {
            this.offsetTable.remove(mq);
            log.info("remove unused mq, {}, {}", mq, this.groupName);
        }
    }
} 

接下来就是通过网络层发送网络请求给Broker进行更新消息对立偏移量。
1.4 读取消息队列的偏移量
两个时刻需要获取Broker保存的偏移量

• 消费者刚启动的时候会去Broker获取消息队列对应的偏移量
• 消费者重平衡后，分配得到新的消息队列，也要重新获取偏移量

readOffset
在DefaultMQPushConsumerImpl的pullMessage方法中，在消费之前会读取一次

 commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY); 

2 服务端的处理逻辑
服务端注册了的消费消息偏移量的请求处理器，首先是有关偏移量的三个请求码
GET_CONSUMER_LIST_BY_GROUP：根据组名获取消费者列表
UPDATE_CONSUMER_OFFSET：更新消费偏移量的请求
QUERY_CONSUMER_OFFSET：查询消费者的偏移量
所以这三个的请求码将交给ConsumerManageProcessor来进行处理。

      /**
         * ConsumerManageProcessor
         */
        ConsumerManageProcessor consumerManageProcessor = new ConsumerManageProcessor(this);
        this.remotingServer.registerProcessor(RequestCode.GET_CONSUMER_LIST_BY_GROUP, consumerManageProcessor, this.consumerManageExecutor);
        this.remotingServer.registerProcessor(RequestCode.UPDATE_CONSUMER_OFFSET, consumerManageProcessor, this.consumerManageExecutor);
        this.remotingServer.registerProcessor(RequestCode.QUERY_CONSUMER_OFFSET, consumerManageProcessor, this.consumerManageExecutor);

2.1 更新消费者传给broker的消费偏移量

内存存储方式
位置：ConsumerOffsetManager的offsetTable中
格式：ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>>,第一层key是主题+消费者组，集群模式下的消费模式；第二层的key是QueueID队列ID。
外部存储位置：

2.2 源码分析
2.2.1 处理偏移量更新请求和更新到内存中的流程
1 请求处理的入口

// RocketMQ里面的通用做法，发送请求时将给请求赋值一个请求码；
// 服务端在接收到请求的时候将根据请求码选择不同的请求处理处理器;
// 统一的接口processRequest()
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
    // ConsuemrManagerProcessor内部又分了不同的处理逻辑
    switch (request.getCode()) {
        // 处理
        case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
            return this.getConsumerListByGroup(ctx, request);
        // 处理更新偏移量
        case RequestCode.UPDATE_CONSUMER_OFFSET:
            return this.updateConsumerOffset(ctx, request);
        case RequestCode.QUERY_CONSUMER_OFFSET:
            return this.queryConsumerOffset(ctx, request);
        default:
            break;
    }
    return null;
}

2 处理更新消费偏移量的入口

private RemotingCommand updateConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
    // 首先创建响应,RocketMQ中惯例做法，具体可参照
    final RemotingCommand response =
            RemotingCommand.createResponseCommand(UpdateConsumerOffsetResponseHeader.class);
    // 解码请求头
    final UpdateConsumerOffsetRequestHeader requestHeader =
            (UpdateConsumerOffsetRequestHeader) request
                    .decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);
    // 调用消费偏移量偏移器进行更新消费偏移量
    this.brokerController.getConsumerOffsetManager()
            .commitOffset(
                    RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                    requestHeader.getConsumerGroup(), // 消费者组
                    requestHeader.getTopic(), // 主题
                    requestHeader.getQueueId(), // 队列ID
                    requestHeader.getCommitOffset()); // 偏移量
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}

3 消费偏移量管理器更新偏移量的入口

public void commitOffset(final String clientHost, final String group, final String topic, final int queueId,
    final long offset) {
    // topic@group
    // 构建key: 主题/消费者组名
    String key = topic + TOPIC_GROUP_SEPARATOR + group;
    this.commitOffset(clientHost, key, queueId, offset);
}

4 将消费者端传上来的消费偏移量存储到内存之中的map

private void commitOffset(final String clientHost, final String key, final int queueId, final long offset) {
    // 使用 主题/消费者名 获取存储偏移量的map<queueId, offset>
    ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
    if (null == map) {
        map = new ConcurrentHashMap<Integer, Long>(32);
        map.put(queueId, offset);
        this.offsetTable.put(key, map);
    } else {
        Long storeOffset = map.put(queueId, offset);
        if (storeOffset != null && offset < storeOffset) {
            log.warn("[NOTIFYME]update consumer offset less than store. clientHost={}, key={}, queueId={}, requestOffset={}, storeOffset={}", clientHost, key, queueId, offset, storeOffset);
        }
    }
}

2.2.2 消息偏移量持久化到磁盘
1、启动定时任务,该定时任务在BrokerController中被启动的；

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            // 持久化偏移量
            BrokerController.this.consumerOffsetManager.persist();
        } catch (Throwable e) {
            log.error("schedule persist consumerOffset error.", e);
        }
    }
}, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

2、调用ConsuemerOffsetManager进行偏移量持久化

public synchronized void persist() {
    // 先进行编码
    String jsonString = this.encode(true);
    if (jsonString != null) {
        // 获取存储文件的路径
        String fileName = this.configFilePath();
        try {
            // 将存储内容存到磁盘
            MixAll.string2File(jsonString, fileName);
        } catch (IOException e) {
            log.error("persist file " + fileName + " exception", e);
        }
    }
}