聊聊rocketmq的ConsumerIdsChangeListener

时间 2020-01-01

标签聊聊 rocketmq consumeridschangelistener 繁體版

原文原文链接

序

本文主要研究一下rocketmq的ConsumerIdsChangeListenerjava

ConsumerGroupEvent

rocketmq-all-4.6.0-source-release/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupEvent.javagit

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.javagithub

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.javaapache

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)ide

doc

ConsumerIdsChangeListener