聊聊rocketmq的ConsumerIdsChangeListener
序
本文主要研究一下rocketmq的ConsumerIdsChangeListener
ConsumerGroupEvent
rocketmq-all-4.6.0-source-release/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupEvent.java
public enum ConsumerGroupEvent { /**
* Some consumers in the group are changed.
*/
CHANGE,
/**
* The group of consumer is unregistered.
*/
UNREGISTER,
/**
* The group of consumer is registered.
*/
REGISTER
}
- ConsumerGroupEvent定义了CHANGE、UNREGISTER、REGISTER三个枚举值
ConsumerIdsChangeListener
rocketmq-all-4.6.0-source-release/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerIdsChangeListener.java
public interface ConsumerIdsChangeListener { void handle(ConsumerGroupEvent event, String group, Object... args);
}
- ConsumerIdsChangeListener定义了handle方法,接收并处理ConsumerGroupEvent
DefaultConsumerIdsChangeListener
rocketmq-all-4.6.0-source-release/broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java
public class DefaultConsumerIdsChangeListener implements ConsumerIdsChangeListener { private final BrokerController brokerController;
public DefaultConsumerIdsChangeListener(BrokerController brokerController) {
this.brokerController = brokerController;
}
@Override
public void handle(ConsumerGroupEvent event, String group, Object... args) {
if (event == null) {
return;
}
switch (event) {
case CHANGE:
if (args == null || args.length < 1) {
return;
}
List<Channel> channels = (List<Channel>) args[0];
if (channels != null && brokerController.getBrokerConfig().isNotifyConsumerIdsChangedEnable()) {
for (Channel chl : channels) {
this.brokerController.getBroker2Client().notifyConsumerIdsChanged(chl, group);
}
}
break;
case UNREGISTER:
this.brokerController.getConsumerFilterManager().unRegister(group);
break;
case REGISTER:
if (args == null || args.length < 1) {
return;
}
Collection<SubscriptionData> subscriptionDataList = (Collection<SubscriptionData>) args[0];
this.brokerController.getConsumerFilterManager().register(group, subscriptionDataList);
break;
default:
throw new RuntimeException("Unknown event " + event);
}
}
}
- DefaultConsumerIdsChangeListener实现了ConsumerIdsChangeListener接口,其handle方法使用switch case来处理ConsumerGroupEvent;对于CHANGE类型执行brokerController.getBroker2Client().notifyConsumerIdsChanged(chl, group);对于UNREGISTER类型执行brokerController.getConsumerFilterManager().unRegister(group);对于REGISTER类型执行brokerController.getConsumerFilterManager().register(group, subscriptionDataList)
小结
ConsumerIdsChangeListener定义了handle方法,接收并处理ConsumerGroupEvent;DefaultConsumerIdsChangeListener实现了ConsumerIdsChangeListener接口,其handle方法使用switch case来处理ConsumerGroupEvent;对于CHANGE类型执行brokerController.getBroker2Client().notifyConsumerIdsChanged(chl, group);对于UNREGISTER类型执行brokerController.getConsumerFilterManager().unRegister(group);对于REGISTER类型执行brokerController.getConsumerFilterManager().register(group, subscriptionDataList)
doc
- ConsumerIdsChangeListener
以上是 聊聊rocketmq的ConsumerIdsChangeListener 的全部内容, 来源链接: utcz.com/z/512151.html