这篇文章主要介绍RocketMQ中broker server之如何实现状态管理,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!
broker server在存储层之上,管理一个broker的状态,通过调用存储层DefaultMessageStore的接口实现消息读写。
broker server相关的核心类如下图:
管理每个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持久化到磁盘中。
管理所有的订阅者组,为每个consumer group维护了以下信息:
topic订阅方式
每个consumer的client channel
Push or Pull消费方式
广播/集群消息模型
维护producer端client channel及其他信息
维护subscription group信息
维护Topic信息
以上是“RocketMQ中broker server之如何实现状态管理”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注亿速云行业资讯频道!
亿速云「云数据库 MySQL」免部署即开即用,比自行安装部署数据库高出1倍以上的性能,双节点冗余防止单节点故障,数据自动定期备份随时恢复。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/zhuhui/blog/4590262