Spring Kafka @KafkaListener源码剖析

在 Spring Boot 中集成 Kafka通过 KafkaListener 注解可快速实现消息消费的编码开发这种通过声明式的方式极大简化了消息监听器的配置提升了开发效率。这里尝试通过分析 KafkaListener 的关键源码弄清楚它背后的原理。一、KafkaListener 处理的入口Bean 后处理器先简单说下 Spring 的 Bean 后处理器——BeanPostProcessor。BeanPostProcessor 是 Spring 中的一个接口它有 2 个方法BeanPostProcessorpostProcessBeforeInitialization 方法会在 Bean 初始化之前被 Spring 调用postProcessAfterInitialization 方法是在 Bean 初始化之后被调用。Spring 在启动过程中会在 Bean 初始化完成后对所有 Bean 进行扫描对实现了 BeanPostProcessor 接口的 Bean 会调用其 postProcessAfterInitialization 方法。二、KafkaListener 注解解析封装为 EndPointKafkaListener 的处理就是在一个 BeanPostProcessor 中——KafkaListenerAnnotationBeanPostProcessor。KafkaListenerAnnotationBeanPostProcessor 在其 postProcessAfterInitialization 方法中会检查当前 bean 上或 bean 中的方法上是否标注了KafkaListener 或KafkaListeners 注解。当检测到注解时KafkaListenerAnnotationBeanPostProcessor 会解析注解属性如 topics、groupId、containerFactory 等并基于这些信息创建 MethodKafkaListenerEndpoint 实例。Endpoint 类封装了消息监听方法的相关元数据包括方法对象、Bean 实例、并发配置和异常处理策略等。随后这些端点endpoint会被包装成 EndpointDescriptor 放入到 KafkaListenerEndpointRegistrar 中。以下是重点部分的源码// KafkaListenerAnnotationBeanPostProcessor public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { // 跳过代码 Class? targetClass AopUtils.getTargetClass(bean); // 查找类上有KafkaListener注解的bean CollectionKafkaListener classLevelListeners this.findListenerAnnotations(targetClass); MapMethod, SetKafkaListener annotatedMethods MethodIntrospector.selectMethods(targetClass, (methodx) - { // 查找方法上有KafkaListener注解的bean SetKafkaListener listenerMethods this.findListenerAnnotations(methodx); return !listenerMethods.isEmpty() ? listenerMethods : null; }); boolean hasClassLevelListeners !classLevelListeners.isEmpty(); boolean hasMethodLevelListeners !annotatedMethods.isEmpty(); if (!hasMethodLevelListeners !hasClassLevelListeners) { this.nonAnnotatedClasses.add(bean.getClass()); this.logger.trace(() - No KafkaListener annotations found on bean type: String.valueOf(bean.getClass())); } else { if (hasMethodLevelListeners) { for(Map.EntryMethod, SetKafkaListener entry : annotatedMethods.entrySet()) { Method method (Method)entry.getKey(); for(KafkaListener listener : (Set)entry.getValue()) { // 解析标记有KafkaListener的方法得到的信息封装为endpoint this.processKafkaListener(listener, method, bean, beanName); } } // 跳过代码 } if (hasClassLevelListeners) { // KafkaListener类标记有KafkaHandler的方法也是监听方法 SetMethod methodsWithHandler MethodIntrospector.selectMethods(targetClass, (methodx) - AnnotationUtils.findAnnotation(methodx, KafkaHandler.class) ! null); if (methodsWithHandler.isEmpty()) { throw new IllegalStateException(No Kafka listener methods in bean: String.valueOf(bean)); } ListMethod multiMethods new ArrayList(methodsWithHandler); // 也是对方法做解析封装为endpoint this.processMultiMethodListeners(classLevelListeners, multiMethods, targetClass, bean, beanName); } } return bean; } protected void processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName) { try { this.globalLock.lock(); Method methodToUse this.checkProxy(method, bean); // 创建endpoint对象用于下面进行封装 MethodKafkaListenerEndpointK, V endpoint new MethodKafkaListenerEndpoint(); endpoint.setMethod(methodToUse); // 解析注解上的属性set到endpoint中 this.processMainAndRetryListeners(kafkaListener, bean, beanName, endpoint, methodToUse, (Class)null); } finally { this.globalLock.unlock(); } } private final KafkaListenerEndpointRegistrar registrar new KafkaListenerEndpointRegistrar(); protected void processListener(MethodKafkaListenerEndpoint?, ? endpoint, KafkaListener kafkaListener, Object bean, String beanName, String[] topics, TopicPartitionOffset[] tps) { this.processKafkaListenerAnnotation(endpoint, kafkaListener, bean, topics, tps); String containerFactory this.resolve(kafkaListener.containerFactory()); KafkaListenerContainerFactory? listenerContainerFactory this.resolveContainerFactory(kafkaListener, containerFactory, beanName); if (listenerContainerFactory instanceof ShareKafkaListenerContainerFactory) { endpoint.setShareConsumer(true); } // 这里将endpoint放入到 KafkaListenerEndpointRegistrar this.registrar.registerEndpoint(endpoint, listenerContainerFactory); }三、创建 ListenerContainerKafkaListenerAnnotationBeanPostProcessor 不仅实现了 BeanPostProcessor 的 postProcessAfterInitialization 方法它还实现了 SmartInitializingSingleton 接口的 afterSingletonsInstantiated 方法。在 Spring 的初始化过程中afterSingletonsInstantiated 是在所有 bean 都执行完 postProcessAfterInitialization 后调用的。这里简单提下 Spring 容器启动过程中的 bean 初始化顺序大致为实例化构造 bean 对象属性填充依赖注入初始化前回调postProcessBeforeInitialization实现 BeanPostProcessor 接口初始化afterPropertiesSet / init-method实现 InitializingBean 接口或PostConstruct 注解的方法初始化后回调postProcessAfterInitialization实现 BeanPostProcessor 接口当所有单例 Bean 都完成了上述 5 步后Spring 才会触发所有单例初始化完成后回调afterSingletonsInstantiated实现 SmartInitializingSingleton 接口上面都执行完成后Spring bean 的初始化过程就完成了后面就会触发启动实现了 SmartLifecycle 接口并且 autoStartup 为 true 的 bean这个流程下面分析源码也会需要用到。所以KafkaListenerAnnotationBeanPostProcessor 的 postProcessAfterInitialization 方法中解析完所有 KafkaListener 类或方法后封装为 endpoint会触发 KafkaListenerAnnotationBeanPostProcessor 的 afterSingletonsInstantiated。这个方法中最重要的一点就是遍历所有 endpoint构造消息监听器容器 ListenerContainer。构造的 MessageListenerContainer 会注册到 KafkaListenerEndpointRegistry 中也就是放入 KafkaListenerEndpointRegistry 的 map 中而 KafkaListenerEndpointRegistry 实现了 SmartLifeCycle 接口所以在注册完成后会触发 KafkaListenerEndpointRegistry 的 start 方法。KafkaListenerEndpointRegistry 的 start 方法中会根据并发度创建多个 KafkaMessageListenerContainer每个 KafkaMessageListenerContainer 会再被调用 start 方法。四、创建 ListenerConsumerKafkaMessageListenerContainer 的 start 方法中会创建 ListenerConsumer。ListenerConsumer 它实际是运行在独立线程中执行 Kafka 消息的拉取和处理循环它的内部涉及对底层 Kafka-client 的 API 的封装。这样整个 KafkaListener 的解析过程基本就清楚了这部分流程的关键代码如下// KafkaListenerAnnotationBeanPostProcessor public void afterSingletonsInstantiated() { // 跳过代码 // registrar虽然实现了InitializingBean接口但它并没有注册为bean所以这里手动调用 // 这个方法的实现是直接调用的registerAllEndpoints方法进行endpoint注册 this.registrar.afterPropertiesSet(); // 跳过代码 } // ------------ // KafkaListenerEndpointRegistrar protected void registerAllEndpoints() { try { this.endpointsLock.lock(); for(KafkaListenerEndpointDescriptor descriptor : this.endpointDescriptors) { // 跳过代码 if (this.endpointRegistry ! null) { // 这里基于endpoint创建MessageListenerContainer // 这里的endpointRegistry就是KafkaListenerEndpointRegistryendpointRegistry实现了SmartLifecycle接口 // 所以会在注册完成后触发KafkaListenerEndpointRegistry的start方法start方法中就是基于MessageListenerContainer创建kafka的ListenerConsumer了 this.endpointRegistry.registerListenerContainer(descriptor.endpoint, this.resolveContainerFactory(descriptor)); } } this.startImmediately true; } finally { this.endpointsLock.unlock(); } } public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory? factory, boolean startImmediately) { // 跳过代码 this.containersLock.lock(); try { Assert.state(!this.listenerContainers.containsKey(id), Another endpoint is already registered with id id ); // 基于endpint创建消息监听器容器然后放入到当前registry的listenerContainers中 MessageListenerContainer container this.createListenerContainer(endpoint, factory); this.listenerContainers.put(id, container); // 跳过代码 } finally { this.containersLock.unlock(); } } // ------------ // KafkaListenerEndpointRegistry public void start() { for (MessageListenerContainer listenerContainer : getListenerContainers()) { startIfNecessary(listenerContainer); } this.running true; } private void startIfNecessary(MessageListenerContainer listenerContainer) { if ((this.contextRefreshed this.alwaysStartAfterRefresh) || listenerContainer.isAutoStartup()) { // 调用MessageListenerContainer 的start方法创建KafkaMessageListenerContainer listenerContainer.start(); } } // ------------ // ConcurrentMessageListenerContainer protected void doStart() { if (!isRunning()) { // 跳过代码 // 根据并发度决定创建多少个KafkaMessageListenerContainer for (int i 0; i this.concurrency; i) { KafkaMessageListenerContainerK, V container constructContainer(containerProperties, topicPartitions, i); configureChildContainer(i, container); if (isPauseRequested()) { container.pause(); } container.start(); // KafkaMessageListenerContainer的start方法中会创建ListenerConsumer this.containers.add(container); } } }五、整个流程的关键类及作用类名作用KafkaListenerAnnotationBeanPostProcessorBean 后处理器扫描 Spring 容器中的 Bean解析带有 KafkaListener 注解的方法并将它们封装为 KafkaListenerEndpoint最终注册到注册中心里面。KafkaListenerEndpointRegistrar端点注册器工具类收集所有 KafkaListenerEndpoint并调用注册中心的注册方法。KafkaListenerEndpointRegistry端点注册中心管理所有监听容器MessageListenerContainer负责创建和启动容器。KafkaListenerContainerFactory监听容器工厂用于创建 MessageListenerContainer 实例。默认实现是 ConcurrentKafkaListenerContainerFactory。MethodKafkaListenerEndpoint封装了 KafkaListener 注解信息如 topics、groupId、配置等以及目标方法最终会基于它创建具体的 MessageListener。MessageListenerContainer消息监听容器接口定义容器的生命周期启动、停止。ConcurrentMessageListenerContainer并发监听容器内部包含多个 KafkaMessageListenerContainer 实例根据 concurrency 参数实现多线程消费。KafkaMessageListenerContainer单线程监听容器负责实际与 Kafka 消费者交互、拉取消息并调用监听器。ListenerConsumerKafkaMessageListenerContainer 的内部类实际运行在独立线程中执行消息拉取和处理的循环。