Kafka-Consumer 源码解析 -- consumer 启动 和 listener 注册和启动
前言
本文主要针对KafkaListener注解的解析和运行过程进行简单的分析，对消费者的启动和listener的注册过程加以说明。
本文不涉及consumer的分区确定和rebalance等问题的说明。
1、KafkaListener注解说明
KafkaListener.java
public @interface KafkaListener {
String id() default "";
String containerFactory() default "";
String[] topics() default {};
String topicPattern() default "";
TopicPartition[] topicPartitions() default {};
String containerGroup() default "";
String errorHandler() default "";
String groupId() default "";
boolean idIsGroup() default true;
String clientIdPrefix() default "";
String beanRef() default "__listener";
String concurrency() default "";
String autoStartup() default "";
String[] properties() default {};
}
常用参数说明：
- id：消费者的id，当GroupId没有被配置的时候，默认id为GroupId
- containerFactory：配置BeanName，listener容器工厂，为
KafkaListenerContainerFactory的实现，spring默认实现类ConcurrentKafkaListenerContainerFactory，用于生成MessageListenerContainer实例，ConcurrentKafkaListenerContainerFactory所对应的MessageListenerContainer实现为ConcurrentMessageListenerContainer。
- topics：需要监听的Topic，可监听多个。
- concurrency：监听当前Topic所运行的线程数，会覆盖
spring.kafka.listener.concurrency配置。当线程数多于Topic 分区数，那么将会有空闲线程存在。
- topicPartitions：可配置更加详细的监听信息，监听某个Topic中的指定分区。同一topic的同一partition确定一个线程，与concurrency存在关系，当配置了topicPartitions且concurrency大于此部分确定的线程数，那么concurrency就不再起作用。
- errorHandler：监听异常处理器，配置BeanName
- groupId：消费组ID，默认为配置文件中配置
spring.kafka.consumer.group-id
2、listener注册
2.1、KafkaListenerAnnotationBeanPostProcessor
KafkaListener解析由KafkaListenerAnnotationBeanPostProcessor完成，这个类实现了BeanPostProcessor接口，这个接口会扫描所有的bean注册。
BeanPostProcessor.java
public interface BeanPostProcessor {
/**
* 在bean初始化之前执行
*/
@Nullable
default Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
/**
* 在bean初始化之后执行
*/
@Nullable
default Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
}
KafkaListenerAnnotationBeanPostProcessor主要实现了postProcessAfterInitialization方法，在每个bean初始化完成之后再进行相应的处理操作，主要为检查bean是否有KafkaListener注解，如果存在，则会执行后续的注册listener的操作。
postProcessAfterInitialization实现：
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
// nonAnnotatedClasses是 类中没有KafkaListener注解标注的方法 的类集合
// 此步主要处理一个类中无KafkaListener方法，且存在多实例，后续的实例便可不操作
if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
Class<?> targetClass = AopUtils.getTargetClass(bean);
// 得到class上的KafkaListener集合
// class上可使用KafkaListener和KafkaListeners注解，这里会将KafkaListeners解析成KafkaListener集合
Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);
final boolean hasClassLevelListeners = classLevelListeners.size() > 0;
final List<Method> multiMethods = new ArrayList<>();
// 得到当前class下所有被KafkaListener和KafkaListeners注解修饰的method，并建立映射关系
Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
(MethodIntrospector.MetadataLookup<Set<KafkaListener>>) method -> {
Set<KafkaListener> listenerMethods = findListenerAnnotations(method);
return (!listenerMethods.isEmpty() ? listenerMethods : null);
});
if (hasClassLevelListeners) {
// 如果类找到KafkaListener，则在当前类中找到被 KafkaHandler 注解修饰的方法集合
// 也就是如果类上被KafkaListener修饰，那么类方法要使用 KafkaHandler 来配合实现
Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass,
(ReflectionUtils.MethodFilter) method ->
AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);
multiMethods.addAll(methodsWithHandler);
}
if (annotatedMethods.isEmpty()) {
// 如果当前类中没有被KafkaListener和KafkaListeners注解修饰的method，将class添加至nonAnnotatedClasses
// 这样做是避免后续这样无解析意义的同class的不同实例的重复解析加载
this.nonAnnotatedClasses.add(bean.getClass());
}
else {
// 遍历annotatedMethods，根据映射关系执行processKafkaListener
for (Map.Entry<Method, Set<KafkaListener>> entry : annotatedMethods.entrySet()) {
Method method = entry.getKey();
for (KafkaListener listener : entry.getValue()) {
processKafkaListener(listener, method, bean, beanName);
}
}
}
if (hasClassLevelListeners) {
// 处理类级别的classLevelListeners和multiMethods映射关系
// processMultiMethodListeners方法内会在处理完成映射关系后，执行使用第一个参数是MethodKafkaListenerEndpoint的processKafkaListener方法重载，上述的processKafkaListener的内部也是使用此重载
processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);
}
}
return bean;
}
processMultiMethodListeners实现：
private void processMultiMethodListeners(Collection<KafkaListener> classLevelListeners, List<Method> multiMethods,
Object bean, String beanName) {
List<Method> checkedMethods = new ArrayList<>();
// 被 KafkaHandler 修饰为默认的方法
Method defaultMethod = null;
for (Method method : multiMethods) {
Method checked = checkProxy(method, bean);
KafkaHandler annotation = AnnotationUtils.findAnnotation(method, KafkaHandler.class);
if (annotation != null && annotation.isDefault()) {
defaultMethod = checked;
}
checkedMethods.add(checked);
}
// 遍历 classLevelListeners，实例化MultiMethodKafkaListenerEndpoint执行注册
for (KafkaListener classLevelListener : classLevelListeners) {
MultiMethodKafkaListenerEndpoint<K, V> endpoint =
new MultiMethodKafkaListenerEndpoint<>(checkedMethods, defaultMethod, bean);
processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
}
}
processKafkaListener -- postProcessAfterInitialization方法中的调用实现：
/**
* 同 processMultiMethodListeners 中的遍历操作一致
* 实例化MethodKafkaListenerEndpoint执行注册
*/
protected void processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName) {
Method methodToUse = checkProxy(method, bean);
MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
endpoint.setMethod(methodToUse);
processListener(endpoint, kafkaListener, bean, methodToUse, beanName);
}
processKafkaListener -- 处理 MethodKafkaListenerEndpoin t执行注册实现：
protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener,
Object bean, Object adminTarget, String beanName) {
String beanRef = kafkaListener.beanRef();
if (StringUtils.hasText(beanRef)) {
this.listenerScope.addListener(beanRef, bean);
}
// 设置 endpoint 属性
endpoint.setBean(bean);
endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
endpoint.setId(getEndpointId(kafkaListener));
endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));
endpoint.setTopicPartitions(resolveTopicPartitions(kafkaListener));
endpoint.setTopics(resolveTopics(kafkaListener));
endpoint.setTopicPattern(resolvePattern(kafkaListener));
endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(), "clientIdPrefix"));
String group = kafkaListener.containerGroup();
if (StringUtils.hasText(group)) {
Object resolvedGroup = resolveExpression(group);
if (resolvedGroup instanceof String) {
endpoint.setGroup((String) resolvedGroup);
}
}
String concurrency = kafkaListener.concurrency();
if (StringUtils.hasText(concurrency)) {
endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));
}
String autoStartup = kafkaListener.autoStartup();
if (StringUtils.hasText(autoStartup)) {
endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));
}
resolveKafkaProperties(endpoint, kafkaListener.properties());
// listener 容器工厂，生产MessageListenerContainer消息监听容器
KafkaListenerContainerFactory<?> factory = null;
String containerFactoryBeanName = resolve(kafkaListener.containerFactory());
if (StringUtils.hasText(containerFactoryBeanName)) {
try {
// 如果手动配置了 containerFactory ，则根据 beanName 从 beanFactory 获取对应实例
factory = this.beanFactory.getBean(containerFactoryBeanName, KafkaListenerContainerFactory.class);
}
catch (NoSuchBeanDefinitionException ex) {
throw new BeanInitializationException();
}
}
endpoint.setBeanFactory(this.beanFactory);
String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
if (StringUtils.hasText(errorHandlerBeanName)) {
endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
}
// 使用 registrar 注册 endpoit，register为KafkaListenerEndpointRegistrar实例
this.registrar.registerEndpoint(endpoint, factory);
if (StringUtils.hasText(beanRef)) {
this.listenerScope.removeListener(beanRef);
}
}
2.2、KafkaListenerEndpointRegistrar
KafkaListenerEndpoint的注册由KafkaListenerEndpointRegistrar完成。2.1节的结尾this.registrar.registerEndpoint(endpoint, factory);是注册调用入口。
registerEndpoint实现：
public void registerEndpoint(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory) {
// 创建 KafkaListenerEndpoint 描述信息对象，并将此对象加入至 endpointDescriptors 中
KafkaListenerEndpointDescriptor descriptor = new KafkaListenerEndpointDescriptor(endpoint, factory);
synchronized (this.endpointDescriptors) {
if (this.startImmediately) { // Register and start immediately
this.endpointRegistry.registerListenerContainer(descriptor.endpoint,
resolveContainerFactory(descriptor), true);
}
else {
this.endpointDescriptors.add(descriptor);
}
}
}
registerEndpoint执行之后，由KafkaListenerAnnotationBeanPostProcessor为入口的解析工作已经完成，但是我们发现registerEndpoint方法只是将descriptor 加入至endpointDescriptors，并没有做一些注册和listener container的创建。
在KafkaListenerEndpointRegistrar类中registerAllEndpoints方法来执行注册所有endpoint的操作。
registerAllEndpoints实现：
protected void registerAllEndpoints() {
synchronized (this.endpointDescriptors) {
// 遍历 endpointDescriptors，使用 KafkaListenerEndpointRegistry 的实例 endpointRegistry 执行listener container 的注册
for (KafkaListenerEndpointDescriptor descriptor : this.endpointDescriptors) {
this.endpointRegistry.registerListenerContainer(
descriptor.endpoint, resolveContainerFactory(descriptor));
}
this.startImmediately = true; // trigger immediate startup
}
}
registerAllEndpoints由afterPropertiesSet调用。afterPropertiesSet是KafkaListenerEndpointRegistrar对InitializingBean接口的实现，实现了InitializingBean接口的bean会在所有属性都完成注入之后，由spring自动调用afterPropertiesSet方法。
但是对于KafkaListenerEndpointRegistrar来说，spring并不会自动调用afterPropertiesSet方法，这是因为KafkaListenerEndpointRegistrar的实例并不是交给spring容器管理的，而是在KafkaListenerAnnotationBeanPostProcessor中new出来的，所以afterPropertiesSet需要手动调用执行。
我们发现在KafkaListenerAnnotationBeanPostProcessor中的afterSingletonsInstantiated方法中调用了registrar.afterPropertiesSet();。
afterSingletonsInstantiated实现：
public void afterSingletonsInstantiated() {
this.registrar.setBeanFactory(this.beanFactory);
if (this.beanFactory instanceof ListableBeanFactory) {
Map<String, KafkaListenerConfigurer> instances =
((ListableBeanFactory) this.beanFactory).getBeansOfType(KafkaListenerConfigurer.class);
for (KafkaListenerConfigurer configurer : instances.values()) {
configurer.configureKafkaListeners(this.registrar);
}
}
// registrar 设置 endpointRegistry
// 在 KafkaListenerEndpointRegistrar 中的方法 registerAllEndpoints 内是使用 KafkaListenerEndpointRegistry 进行注册操作
if (this.registrar.getEndpointRegistry() == null) {
if (this.endpointRegistry == null) {
this.endpointRegistry = this.beanFactory.getBean(
KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
KafkaListenerEndpointRegistry.class);
}
this.registrar.setEndpointRegistry(this.endpointRegistry);
}
if (this.defaultContainerFactoryBeanName != null) {
this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);
}
// MessageHandlerMethodFactory 用于创建 InvocableHandlerMethod
// InvocableHandlerMethod 是 listener 对应的 执行器，也就是 在拉取到数据之后，会调用 HandlerAdapter 的对应方法，执行消费的过程（执行KafkaListener注解修饰的方法）
// InvocableHandlerMethod 的执行过程是借助 HandlerAdapter 实现
MessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory();
if (handlerMethodFactory != null) {
this.messageHandlerMethodFactory.setHandlerMethodFactory(handlerMethodFactory);
}
else {
addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService);
}
// 调用 registrar.afterPropertiesSet() 执行注册过程
this.registrar.afterPropertiesSet();
}
afterSingletonsInstantiated是对接口SmartInitializingSingleton的实现。
SmartInitializingSingleton会在spring容器加载过所有bean之后自动调用，前提是实现这个接口的bean交予spring管理，KafkaListenerAnnotationBeanPostProcessor满足此条件，也是说在spring所有bean加载完成之后会自动调用KafkaListenerEndpointRegistrar的registerAllEndpoints。
KafkaListenerEndpointRegistrar的registerAllEndpoints是使用KafkaListenerEndpointRegistry的registerListenerContainer方法完成注册。
2.3、KafkaListenerEndpointRegistry
registerListenerContainer实现：
public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory) {
registerListenerContainer(endpoint, factory, false);
}
public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory,
boolean startImmediately) {
String id = endpoint.getId();
synchronized (this.listenerContainers) {
// 根据 endpoint 创建 message listener 容器
// message listener 容器 用于承载 消息的监听和消费
// 这里创建的是 ConcurrentMessageListenerContainer 实例，是因为createListenerContainer中使用ConcurrentKafkaListenerContainerFactory进行创建container
MessageListenerContainer container = createListenerContainer(endpoint, factory);
// 保存 container ，用于后续的启动
this.listenerContainers.put(id, container);
if (StringUtils.hasText(endpoint.getGroup()) && this.applicationContext != null) {
List<MessageListenerContainer> containerGroup;
if (this.applicationContext.containsBean(endpoint.getGroup())) {
containerGroup = this.applicationContext.getBean(endpoint.getGroup(), List.class);
}
else {
containerGroup = new ArrayList<MessageListenerContainer>();
this.applicationContext.getBeanFactory().registerSingleton(endpoint.getGroup(), containerGroup);
}
containerGroup.add(container);
}
// 查看调用此方法的参数，startImmediately为 false
// 所以此步并不会执行 start
if (startImmediately) {
startIfNecessary(container);
}
}
}
查看上述代码，registerListenerContainer在注册完成后并不会启动容器的内容，实际的启动交给 SmartLifecycle接口处理，这里KafkaListenerEndpointRegistry类实现了此接口，会在项目所有bean加载和初始化完毕执行start方法。
start实现：
public void start() {
// 遍历容器，执行start
for (MessageListenerContainer listenerContainer : getListenerContainers()) {
startIfNecessary(listenerContainer);
}
this.running = true;
}
private void startIfNecessary(MessageListenerContainer listenerContainer) {
if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
// 执行 listener 容器的 start
// 这里的 listenerContainer 是 ConcurrentMessageListenerContainer 实例
listenerContainer.start();
}
}
下面查看 ConcurrentMessageListenerContainer的 start方法。
2.4、ConcurrentMessageListenerContainer
start实现：
// 此方法在父类AbstractMessageListenerContainer中
public final void start() {
checkGroupId();
synchronized (this.lifecycleMonitor) {
if (!isRunning()) {
doStart();
}
}
}
protected void doStart() {
if (!isRunning()) {
checkTopics();
ContainerProperties containerProperties = getContainerProperties();
// 这里的 topicPartitions 为注解配置
TopicPartitionOffset[] topicPartitions = containerProperties.getTopicPartitionsToAssign();
// 如果配置了 topicPartitions ，那么为防止出现空闲线程，做以下的处理
// 如果说 topic 实际的 partition 数 小于处理之后的 this.concurrency，还是会出现空闲线程
if (topicPartitions != null && this.concurrency > topicPartitions.length) {
this.concurrency = topicPartitions.length;
}
setRunning(true);
// 循环创建 concurrency 个 KafkaMessageListenerContainer 用于处理 当前kafka的数据
// KafkaMessageListenerContainer 可以看作一个线程（实际线程为内部类ListenerConsumer实现）
for (int i = 0; i < this.concurrency; i++) {
KafkaMessageListenerContainer<K, V> container;
if (topicPartitions == null) {
container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, containerProperties);
}
else {
container = new KafkaMessageListenerContainer<>(this, this.consumerFactory,
containerProperties, partitionSubset(containerProperties, i));
}
String beanName = getBeanName();
container.setBeanName((beanName != null ? beanName : "consumer") + "-" + i);
container.setApplicationContext(getApplicationContext());
if (getApplicationEventPublisher() != null) {
container.setApplicationEventPublisher(getApplicationEventPublisher());
}
container.setClientIdSuffix("-" + i);
container.setGenericErrorHandler(getGenericErrorHandler());
container.setAfterRollbackProcessor(getAfterRollbackProcessor());
container.setRecordInterceptor(getRecordInterceptor());
container.setEmergencyStop(() -> {
stop(() -> {});
publishContainerStoppedEvent();
});
if (isPaused()) {
container.pause();
}
// 启动 KafkaMessageListenerContainer 容器
container.start();
this.containers.add(container);
}
}
}
下面看 KafkaMessageListenerContainer的start实现
2.5、KafkaMessageListenerContainer
start实现：
// 此方法在父类AbstractMessageListenerContainer中
public final void start() {
checkGroupId();
synchronized (this.lifecycleMonitor) {
if (!isRunning()) {
Assert.state(this.containerProperties.getMessageListener() instanceof GenericMessageListener,
() -> "A " + GenericMessageListener.class.getName() + " implementation must be provided");
doStart();
}
}
}
// doStart 执行过后，整个 listener 的注册动作完成
protected void doStart() {
if (isRunning()) {
return;
}
if (this.clientIdSuffix == null) { // stand-alone container
checkTopics();
}
ContainerProperties containerProperties = getContainerProperties();
checkAckMode(containerProperties);
Object messageListener = containerProperties.getMessageListener();
// 设置 TaskExecutor 用于启动 ListenerConsumer ，ListenerConsumer是实际拉取和消费数据的执行
// ListenerConsumer 的执行为异步
if (containerProperties.getConsumerTaskExecutor() == null) {
SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor(
(getBeanName() == null ? "" : getBeanName()) + "-C-");
containerProperties.setConsumerTaskExecutor(consumerExecutor);
}
GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;
ListenerType listenerType = determineListenerType(listener);
this.listenerConsumer = new ListenerConsumer(listener, listenerType);
setRunning(true);
this.startLatch = new CountDownLatch(1);
// 使用 SimpleAsyncTaskExecutor 开启异步线程 执行 this.listenerConsumer，处理数据的拉取和消费过程
this.listenerConsumerFuture = containerProperties
.getConsumerTaskExecutor()
.submitListenable(this.listenerConsumer);
try {
if (!this.startLatch.await(containerProperties.getConsumerStartTimout().toMillis(), TimeUnit.MILLISECONDS)) {
publishConsumerFailedToStart();
}
}catch (@SuppressWarnings(UNUSED) InterruptedException e) {
Thread.currentThread().interrupt();
}
}
3、总结
注册 listener 过程：
- 使用 KafkaListenerAnnotationBeanPostProcessor 扫描所有bean，判断是否含有 KafkaListener 注解。
- 将扫描到 bean 和 method 封装，保存至 KafkaListenerEndpointRegistrar 的 endpointDescriptors。
- 借助 KafkaListenerAnnotationBeanPostProcessor 实现 InitializingBean 接口，对 endpointDescriptors 内的 endpoint 执行 KafkaListenerEndpointRegistry 的 registerListenerContainer 方法进行注册。
- KafkaListenerEndpointRegistry 的 registerListenerContainer 方法将 endpoint 封装 为 ConcurrentMessageListenerContainer存放至 containerGroup 中。
- 借助 KafkaListenerEndpointRegistry 实现 SmartLifecycle 接口，完成对 containerGroup 中所有 ConcurrentMessageListenerContainer 的 start。
- ConcurrentMessageListenerContainer 的 start 完成对当前 KafkaListener 所需实际处理对象 KafkaMessageListenerContainer 的实例化，这里的 KafkaMessageListenerContainer 可以有多个，由配置决定。每个 KafkaMessageListenerContainer 会开一个线程处理消息的拉取和消费。当前 KafkaListener 对应的 concurrency 大于 topics 的 partition，会出现空闲线程，concurrency 大于 Kafka Listener 配置的 topicPartitions ，会使用 topicPartitions 的大小确认实例化的 KafkaMessageListenerContainer 个数。
- KafkaMessageListenerContainer 的 start 完成对 ListenerConsumer 实例化和启动。ListenerConsumer 是实际处理消息的类，为异步处理。
