Spring Kafka源码阅读小记之消费者

@Slf4j
@Component
@KafkaListener(topics = "test")
public class DemoClassConsumer {

    @KafkaHandler(isDefault = true)
    public void handle(String payload) {
        log.info("default handle - receive data: {}", payload);
    }

    @KafkaHandler
    public void handleString(String payload) {
        log.info("string handle - receive data: {}", payload);
    }
}

@Slf4j
@Component
public class DemoConsumer {

    @KafkaListener(topics = "test")
    public void handle(String message) {
        log.info("receive message: {}", message);
    }
}

KafkaListener(topics = "test-consumer", autoStartup = "false")

@Slf4j
@Service
@AllArgsConstructor
public class ConsumerManager implements ApplicationRunner {
    private KafkaListenerEndpointRegistry endpointRegistry;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (getDistributeLockSuccessfully()) {
            endpointRegistry.start(); //此方法会将没运行的消费者全部打开
        }
    }
}

private void invokeOnMessage(final ConsumerRecord<K, V> record, @SuppressWarnings(RAWTYPES) @Nullable Producer producer) {
    // (1)检查key和value是否是一个DeserializationException异常
    // (2)如果key和value为空，判断其headers中是否包含Spring反序列化异常
    // (3)调用实际的消息处理逻辑
    doInvokeOnMessage(record);
    
    // (4)commit offset
    ackCurrent(record, producer);
}

private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> record,@SuppressWarnings(RAW_TYPES) Producer producer,Iterator<ConsumerRecord<K, V>> iterator) {
    try {
        // (1) 
        invokeOnMessage(record, producer);
    } catch (RuntimeException e) {
        // (2)
        if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {
            ackCurrent(record);
        }

        // (3)各种异常处理balabala
        ...
    }
    return null;
}

public boolean isAckOnError() {
    return this.ackOnError &&
                !(AckMode.MANUAL_IMMEDIATE.equals(this.ackMode)     
                ||AckMode.MANUAL.equals(this.ackMode));
}

@Configuration
public class KafkaBootstrapConfiguration {

    @SuppressWarnings("rawtypes")
    @Bean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public KafkaListenerAnnotationBeanPostProcessor kafkaListenerAnnotationProcessor() {
        return new KafkaListenerAnnotationBeanPostProcessor();
    }

    @Bean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
    public KafkaListenerEndpointRegistry defaultKafkaListenerEndpointRegistry() {
        return new KafkaListenerEndpointRegistry();
    }

}

/**
 * Bean post-processor that registers methods annotated with {@link KafkaListener}
 * to be invoked by a Kafka message listener container created under the covers
 * by a {@link org.springframework.kafka.config.KafkaListenerContainerFactory}
 * according to the parameters of the annotation.
 */

/**
 * Creates the necessary {@link MessageListenerContainer} instances for the
 * registered {@linkplain KafkaListenerEndpoint endpoints}. Also manages the
 * lifecycle of the listener containers, in particular within the lifecycle
 * of the application context.
 *
 * <p>Contrary to {@link MessageListenerContainer}s created manually, listener
 * containers managed by registry are not beans in the application context and
 * are not candidates for autowiring. Use {@link #getListenerContainers()} if
 * you need to access this registry's listener containers for management purposes.
 * If you need to access to a specific message listener container, use
 * {@link #getListenerContainer(String)} with the id of the endpoint.
 */

@Override
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
    //一个巨大的判断
    if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
        // 获取原始Bean类型，以防被代理
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        //开始干实事了，找类级别的@KafkaListener注解
        Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);

        final boolean hasClassLevelListeners = classLevelListeners.size() > 0;
        final List<Method> multiMethods = new ArrayList<>();

        //又干了件实事，找方法级别的@KafkaListener注解
        Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,(MethodIntrospector.MetadataLookup<Set<KafkaListener>>) method -> {
                        Set<KafkaListener> listenerMethods = findListenerAnnotations(method);
                        return (!listenerMethods.isEmpty() ? listenerMethods : null);
        });

        //如果有类级别的注解，则还需要确定找到其中的@KafkaHandler注解，是不是和第一章的使用方法对上了
        if (hasClassLevelListeners) {
            Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass,(ReflectionUtils.MethodFilter) method -> AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);
                multiMethods.addAll(methodsWithHandler);
        }

        if (annotatedMethods.isEmpty()) {
            this.nonAnnotatedClasses.add(bean.getClass());
            if (this.logger.isTraceEnabled()) {
                this.logger.trace("No @KafkaListener annotations found on bean type: " + bean.getClass());
            }
        } else { //如果找到了方法级别的注解，丢给别人再去包装
            for (Map.Entry<Method, Set<KafkaListener>> entry : annotatedMethods.entrySet()) {
                Method method = entry.getKey();
                for (KafkaListener listener : entry.getValue()) {
                    //（1）包装方法级别的消费者
                    processKafkaListener(listener, method, bean, beanName);
                }
            }

        //如果有类级别的注解，也丢给别人再去包装。
        if (hasClassLevelListeners) {
            //（2）包装类级别的消费则
            processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);
        }
    }
    return bean;
}

public void registerEndpoint(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory) {
  // 参数校验
  ...

  // 打包endpoint和其对应的工厂，方便存储
  KafkaListenerEndpointDescriptor descriptor = new KafkaListenerEndpointDescriptor(endpoint, factory);

  // 同步一下ArrayList，防止被同时操作
  synchronized (this.endpointDescriptors) {
    if (this.startImmediately) {
      this.endpointRegistry.registerListenerContainer(descriptor.endpoint, resolveContainerFactory(descriptor), true);
    } else {
      this.endpointDescriptors.add(descriptor);
    }
  }
}

@Override
public void afterSingletonsInstantiated() {
    // 设置一些东西
    ...

    // 为某个属性设置EndpointRegistry，眼熟吧
    if (this.registrar.getEndpointRegistry() == null) {
        if (this.endpointRegistry == null) {
            Assert.state(this.beanFactory != null,
                        "BeanFactory must be set to find endpoint registry by bean name");
            this.endpointRegistry = this.beanFactory.getBean(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,KafkaListenerEndpointRegistry.class);
        }

        // 实际上这个endpointRegistry的Bean就是在自动配置中声明的那个
        this.registrar.setEndpointRegistry(this.endpointRegistry);
    }

    // Actually register all listeners
    this.registrar.afterPropertiesSet();
}

@Override
public void afterPropertiesSet() {
    registerAllEndpoints();
}

protected void registerAllEndpoints() {
    synchronized (this.endpointDescriptors) {
        for (KafkaListenerEndpointDescriptor descriptor : this.endpointDescriptors) {
            this.endpointRegistry.registerListenerContainer( descriptor.endpoint, resolveContainerFactory(descriptor));
        }
    this.startImmediately = true;  // trigger immediate startup
    }
}

public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory) {
    registerListenerContainer(endpoint, factory, false);
}

public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory, boolean startImmediately) {
    // 参数校验
    ...

    // 串一下
    synchronized (this.listenerContainers) {
        Assert.state(!this.listenerContainers.containsKey(id), "Another endpoint is already registered with id '" + id + "'");
    //（1）包装之旅还没有结束...
        MessageListenerContainer container = createListenerContainer(endpoint, factory);
        this.listenerContainers.put(id, container);

        if (startImmediately) {
            startIfNecessary(container);
        }
    }
}

@Override
protected void doStart() {
    // 参数校验和设置
    ...

    for (int i = 0; i < this.concurrency; i++) {
        //（1）根据concurrency数量，每一次循环又单独初始化一个container
        KafkaMessageListenerContainer<K, V> container;
        ...
        container.start();
        this.containers.add(container);
    }
}

protected void doStart() {
    // 校验参数之类的
    ...

    // Runnable
    this.listenerConsumer = new ListenerConsumer(listener, listenerType);

    // 另起线程执行了
    this.listenerConsumerFuture = containerProperties
            .getConsumerTaskExecutor()
            .submitListenable(this.listenerConsumer);
}

public void run() {
    //定义消费起点
    ...
    // 持续拉取和消费
    while (isRunning()) {
        ...
        pollAndInvoke();
        ...
    }
    ...
}

protected void pollAndInvoke() {
    // 提交offset ...
    // 检查是否需要重新定义消费起点 ...
    // 检查是否被暂停 ...
    // 拉取消息
    ConsumerRecords<K, V> records = this.consumer.poll(this.pollTimeout);

    // 消费消息
    invokeListener(records);

    //...
}

graph TD;
 A("@KafkaListener: 注解参数")-->B("MethodKafkaListenerEndpoint");
 B-->D;
 C("ConcurrentKafkaListenerContainerFactory: 配置参数")-->D("KafkaListenerEndpointDescriptor");
 D-.->|注册|E("KafkaListenerEndpointRegistrar: 辅助类");
 E-->F("ConcurrentMessageListenerContainer");
 F-.->|注册|G("KafkaListenerEndpointRegistry");
 G-.->|start|H("ConcurrentMessageListenerContainer");
 H-->I("KafkaMessageListenerContainer1");
 H-->M("KafkaMessageListenerContainer...");
 I-->N("ListenerConsumer: Runnable");
 N-->N;
 J("Spring BeanPostProcessor")-.->|驱动|A
 K("Spring InitializingBean")-.->|驱动|E
 L("Spring SmartLifeCycle")-.->|驱动|G