淮安淮阴网站建设,汽车之家app下载,昆山普立斯特做的有网站,锦州企业网站建设前言
SubscriptionGroupManager 继承了ConfigManager配置管理组件#xff0c;拥有将内存数据持久化到磁盘文件subscriptionGroup.json的能力。它主要负责维护所有消费组在内存中的订阅数据。 源码版本#xff1a;4.9.3
源码架构图 核心数据结构
主要的数据结构比较简单拥有将内存数据持久化到磁盘文件subscriptionGroup.json的能力。它主要负责维护所有消费组在内存中的订阅数据。 源码版本4.9.3
源码架构图 核心数据结构
主要的数据结构比较简单维护了Map消费组名称, 订阅组配置的映射关系。
// 订阅组管理组件
public class SubscriptionGroupManager extends ConfigManager {private static final InternalLogger log InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);// Map消费组名称订阅组配置private final ConcurrentMapString, SubscriptionGroupConfig subscriptionGroupTable new ConcurrentHashMapString, SubscriptionGroupConfig(1024);// 内存数据版本号private final DataVersion dataVersion new DataVersion();
}
深入看下SubscriptionGroupConfig 的数据结构。
public class SubscriptionGroupConfig {// 消费组名称private String groupName;// 是否开启消费private boolean consumeEnable true;// 是否允许消费最早消息private boolean consumeFromMinEnable true;// 是否允许广播消费private boolean consumeBroadcastEnable true;// 重试队列数private int retryQueueNums 1;// 重试最大次数private int retryMaxTimes 16;// brokerIdprivate long brokerId MixAll.MASTER_ID;// 当产生慢消费时选择第几个brokerprivate long whichBrokerWhenConsumeSlowly 1;// 是否通知消费者ids变化private boolean notifyConsumerIdsChangedEnable true;
} 核心数据行为
数据行为主要都是对上面提到的数据结构的维护代码 注释如下
// 订阅组管理组件
public class SubscriptionGroupManager extends ConfigManager {public SubscriptionGroupManager() {this.init();}public SubscriptionGroupManager(BrokerController brokerController) {this.brokerController brokerController;this.init();}private void init() {{// 初始化系统消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.TOOLS_CONSUMER_GROUP);this.subscriptionGroupTable.put(MixAll.TOOLS_CONSUMER_GROUP, subscriptionGroupConfig);}{// 初始化过滤服务消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.FILTERSRV_CONSUMER_GROUP);this.subscriptionGroupTable.put(MixAll.FILTERSRV_CONSUMER_GROUP, subscriptionGroupConfig);}{// 初始化自测消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.SELF_TEST_CONSUMER_GROUP);this.subscriptionGroupTable.put(MixAll.SELF_TEST_CONSUMER_GROUP, subscriptionGroupConfig);}{// 初始化http代理消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.ONS_HTTP_PROXY_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true);this.subscriptionGroupTable.put(MixAll.ONS_HTTP_PROXY_GROUP, subscriptionGroupConfig);}{// 初始化ONS_API_PULL消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.CID_ONSAPI_PULL_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true); // 激活广播模式this.subscriptionGroupTable.put(MixAll.CID_ONSAPI_PULL_GROUP, subscriptionGroupConfig);}{// 初始化ONS_API_PERMISSION消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.CID_ONSAPI_PERMISSION_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true);this.subscriptionGroupTable.put(MixAll.CID_ONSAPI_PERMISSION_GROUP, subscriptionGroupConfig);}{// 初始化ONS_API_OWNER消费组SubscriptionGroupConfig subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(MixAll.CID_ONSAPI_OWNER_GROUP);subscriptionGroupConfig.setConsumeBroadcastEnable(true);this.subscriptionGroupTable.put(MixAll.CID_ONSAPI_OWNER_GROUP, subscriptionGroupConfig);}}// 更新订阅配置且更新内存数据版本号public void updateSubscriptionGroupConfig(final SubscriptionGroupConfig config) {SubscriptionGroupConfig old this.subscriptionGroupTable.put(config.getGroupName(), config);if (old ! null) {log.info(update subscription group config, old: {} new: {}, old, config);} else {log.info(create new subscription group, {}, config);}this.dataVersion.nextVersion();this.persist();}// 失效消费组public void disableConsume(final String groupName) {SubscriptionGroupConfig old this.subscriptionGroupTable.get(groupName);if (old ! null) {old.setConsumeEnable(false);this.dataVersion.nextVersion();}}// 查找指定消费组的订阅配置public SubscriptionGroupConfig findSubscriptionGroupConfig(final String group) {SubscriptionGroupConfig subscriptionGroupConfig this.subscriptionGroupTable.get(group);if (null subscriptionGroupConfig) {if (brokerController.getBrokerConfig().isAutoCreateSubscriptionGroup() || MixAll.isSysConsumerGroup(group)) {subscriptionGroupConfig new SubscriptionGroupConfig();subscriptionGroupConfig.setGroupName(group);SubscriptionGroupConfig preConfig this.subscriptionGroupTable.putIfAbsent(group, subscriptionGroupConfig);if (null preConfig) {log.info(auto create a subscription group, {}, subscriptionGroupConfig.toString());}this.dataVersion.nextVersion();this.persist();}}return subscriptionGroupConfig;}// 将内存数据结构编码成字符串Overridepublic String encode() {return this.encode(false);}// 获取配置文件路径Overridepublic String configFilePath() {return BrokerPathConfigHelper.getSubscriptionGroupPath(this.brokerController.getMessageStoreConfig().getStorePathRootDir());}// 从字符串中恢复数据写回内存数据结构Overridepublic void decode(String jsonString) {if (jsonString ! null) {SubscriptionGroupManager obj RemotingSerializable.fromJson(jsonString, SubscriptionGroupManager.class);if (obj ! null) {this.subscriptionGroupTable.putAll(obj.subscriptionGroupTable);this.dataVersion.assignNewOne(obj.dataVersion);this.printLoadDataWhenFirstBoot(obj);}}}// 将内存数据结构编码成字符串public String encode(final boolean prettyFormat) {return RemotingSerializable.toJson(this, prettyFormat);}// 当第一次启动时打印加载数据时的日志private void printLoadDataWhenFirstBoot(final SubscriptionGroupManager sgm) {IteratorEntryString, SubscriptionGroupConfig it sgm.getSubscriptionGroupTable().entrySet().iterator();while (it.hasNext()) {EntryString, SubscriptionGroupConfig next it.next();log.info(load exist subscription group, {}, next.getValue().toString());}}public ConcurrentMapString, SubscriptionGroupConfig getSubscriptionGroupTable() {return subscriptionGroupTable;}public DataVersion getDataVersion() {return dataVersion;}// 删除指定消费组的订阅配置public void deleteSubscriptionGroupConfig(final String groupName) {SubscriptionGroupConfig old this.subscriptionGroupTable.remove(groupName);if (old ! null) {log.info(delete subscription group OK, subscription group:{}, old);this.dataVersion.nextVersion();this.persist();} else {log.warn(delete subscription group failed, subscription groupName: {} not exist, groupName);}}
}
