我理解的RocketMQ—消息偏移量管理分析

1 客户端逻辑

1.1 概述

偏移量管理主要是指管理每个消息队列的消费进度：集群模式消费下会将消息队列的消费进度保存在Broker端，广播模式消费下消息队列的消费进度保存在消费者本地。

组件分析：RocketMQ定义了一个接口OffsetStore。它的实现类有两个：

RemoteBrokerOffsetStore

和

LocalFileOffsetStore

前者主要是集群消费模式下使用，即与broker进行打交道，将消息队列的消费偏移量通过网络传递给Broker；后者主要是广播消费模式下使用，即直接将消费偏移量存储在消费者所在的本地中。入下图所示：

我理解的RocketMQ—消息偏移量管理分析

offsetstore

保存在消费者内部客户单

ConsumerInner

的实现类中的，其初始化创建的时机在内部客户端的

start()

方法中。

switch (this.defaultMQPushConsumer.getMessageModel()) {
    // 广播模式偏移量持久化为本地
    case BROADCASTING:
        this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
        break;
    // 集群模式下，偏移量持久化方式为远程
    case CLUSTERING:
        this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
        break;
    default:
        break;
}

下面主要分析

RemoteBrokerOffsetStore

的逻辑。

主要是两个逻辑，如下图所示

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

我理解的RocketMQ—消息偏移量管理分析

1.2 更新消息队列的偏移量

updateOffset

并发消息消费服务中

ConsumeMessageConcurrentlyService

processConsumeResult()

处理消息消费结果的方法中在消息处理完成以后会调用更新消息队列的偏移量

// 获取偏移量存储实现，并调用其更新偏移量方法更新偏移量

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方法中

在消费之前会读取一次

2 服务端的处理逻辑

服务端注册了的消费消息偏移量的请求处理器，首先是有关偏移量的三个请求码

GET_CONSUMER_LIST_BY_GROUP：根据组名获取消费者列表
UPDATE_CONSUMER_OFFSET：更新消费偏移量的请求
QUERY_CONSUMER_OFFSET：查询消费者的偏移量

所以这三个的请求码将交给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, ConcurrentMap<Integer, Long>> ,第一层 key 是主题+消费者组，集群模式下的消费模式；第二层的 key 是 QueueID 队列 ID 。

外部存储位置：

下图表示整个更新消费偏移量和持久化的过程；整体流程: 先更新内存中的存储offetTable，然后通过一个持久化的线程将offsetTable中的数据落盘持久化。

我理解的RocketMQ—消息偏移量管理分析

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) {
    // [email protected]
    // 构建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);
        }
    }
}

我理解的RocketMQ—消息偏移量管理分析

1 客户端逻辑

1.1 概述

1.2 更新消息队列的偏移量

1.3 向Broker发送消息偏移量

1.4 读取消息队列的偏移量

2 服务端的处理逻辑

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

2.2 源码分析

2.2.1 处理偏移量更新请求和更新到内存中的流程

2.2.2 消息偏移量持久化到磁盘

继续阅读

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

express 搭建后台构建工程目录结构

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method