这篇文章主要介绍RocketMQ中broker server之如何实现状态管理,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!
创新互联建站专注于企业营销型网站建设、网站重做改版、临湘网站定制设计、自适应品牌网站建设、H5网站设计、购物商城网站建设、集团公司官网建设、成都外贸网站制作、高端网站制作、响应式网页设计等建站业务,价格优惠性价比高,为临湘等各大城市提供网站开发制作服务。
broker server在存储层之上,管理一个broker的状态,通过调用存储层DefaultMessageStore的接口实现消息读写。
broker server相关的核心类如下图:
管理每个consumer消费进度
public class ConsumerOffsetManager extends ConfigManager { private ConcurrentMap> offsetTable = new ConcurrentHashMap >(512); private void commitOffset(final String clientHost, final String key, final int queueId, final long offset) { ConcurrentMap map = this.offsetTable.get(key); if (null == map) { map = new ConcurrentHashMap (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之如何实现状态管理”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注创新互联行业资讯频道!
售后响应及时
7×24小时客服热线数据备份
更安全、更高效、更稳定价格公道精准
项目经理精准报价不弄虚作假合作无风险
重合同讲信誉,无效全额退款