RocketMQ中broker server之如何實現狀態管理

這篇文章主要介紹RocketMQ中broker server之如何實現狀態管理，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

broker server在存儲層之上，管理一個broker的狀態，通過調用存儲層DefaultMessageStore的接口實現消息讀寫。

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之如何實現狀態管理”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業資訊頻道！