RocketMQ中broker server之如何实现状态管理

这篇文章主要介绍RocketMQ中broker server之如何实现状态管理,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!

broker server在存储层之上,管理一个broker的状态,通过调用存储层DefaultMessageStore的接口实现消息读写。

broker server相关的核心类如下图:

RocketMQ中broker server之如何实现状态管理

一、ConsumerOffsetManager

管理每个consumer消费进度

public class ConsumerOffsetManager extends ConfigManager {

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

    private void commitOffset(final String clientHost, final String key, final int queueId, final long 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);
            }
        }
    }

    // ...
}

ConsumerOffsetManager会周期性把offsetTable持久化到磁盘中。

二、ConsumerManager

管理所有的订阅者组,为每个consumer group维护了以下信息:

  • topic订阅方式

  • 每个consumer的client channel

  • Push or Pull消费方式

  • 广播/集群消息模型

三、ProducerManager

维护producer端client channel及其他信息

四、SubscriptionGroupManager

维护subscription group信息

五、TopicConfigManager

维护Topic信息

以上是“RocketMQ中broker server之如何实现状态管理”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注亿速云行业资讯频道!

原创文章,作者:306829225,如若转载,请注明出处:https://blog.ytso.com/222862.html

(0)
上一篇 2022年1月6日
下一篇 2022年1月6日

相关推荐

发表回复

登录后才能评论