深入理解 Spring Kafka 监听器容器与空闲事件：暂停状态的影响

/**
 * Return true if the consumer was paused at the time the idle event was published.
 * @return paused.
 * @since 2.1.5
 */
public boolean isPaused() {
    return this.paused;
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import java.time.Duration;

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 设置空闲事件间隔为10秒
        factory.getContainerProperties().setIdleEventInterval(Duration.ofSeconds(10).toMillis());
        return factory;
    }
}

import org.springframework.context.event.EventListener;
import org.springframework.kafka.event.ListenerContainerIdleEvent;
import org.springframework.stereotype.Component;

@Component
public class KafkaIdleEventListener {

    @EventListener
    public void handleIdleEvent(ListenerContainerIdleEvent event) {
        String listenerId = event.getContainer().getListenerId();
        long idleTime = event.getIdleTime();
        boolean isPaused = event.isPaused();

        if (isPaused) {
            System.out.printf("Listener container '%s' is idle (paused) for %d ms. No messages processed while paused.%n", listenerId, idleTime);
            // 容器处于暂停状态下的空闲，可能不需要触发紧急报警，但可以记录日志
            // 示例：可以更新监控指标，表明容器处于预期内的非活跃状态
        } else {
            System.out.printf("Listener container '%s' is idle (active) for %d ms. No new messages received.%n", listenerId, idleTime);
            // 容器处于运行状态下的空闲，可能意味着没有新消息，需要进一步调查或触发报警
            // 示例：发送邮件或短信通知，检查上游生产者状态
        }
    }
}