聊聊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

回到顶部