如何在生产者无消息时停止Spring Kafka消费者及KafkaListenerContainerFactory

要实现这个需求，核心思路是让生产者在发送完所有业务消息后发送一个结束标记信号，消费者识别到该信号后停止对应的监听容器；如果需要彻底关闭应用，还可以进一步触发Spring上下文的关闭。下面是具体的实现方案：

1. 生产者：发送结束标记 #

当数据库的所有记录都发送到Kafka后，发送一个约定好的特殊消息（比如"END_OF_STREAM"）作为结束信号。为了确保所有业务消息都成功发送到Kafka后再发送结束标记，可以同步等待发送结果：

@Service
public class DbKafkaProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private static final String TARGET_TOPIC = "your-business-topic";
    // 如果是并发消费，这里设置和消费者并发数一致的结束消息数量
    private static final int CONCURRENCY = 2;

    public DbKafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendAllDbRecords() throws InterruptedException, ExecutionException {
        // 1. 从数据库拉取所有待发送记录
        List<String> dbRecords = fetchRecordsFromDatabase();

        // 2. 同步发送所有业务记录，确保消息落地Kafka
        for (String record : dbRecords) {
            kafkaTemplate.send(TARGET_TOPIC, record)
                    .get(10, TimeUnit.SECONDS); // 等待发送确认，超时时间可根据业务调整
        }

        // 3. 发送结束标记：如果是并发消费，发送与并发数相同数量的结束消息
        // 保证每个消费者实例都能收到一个结束信号
        for (int i = 0; i < CONCURRENCY; i++) {
            kafkaTemplate.send(TARGET_TOPIC, "END_OF_STREAM")
                    .get(10, TimeUnit.SECONDS);
        }

        System.out.println("All records and end markers sent successfully");
    }

    private List<String> fetchRecordsFromDatabase() {
        // 替换为实际的数据库查询逻辑
        return Arrays.asList("record-1", "record-2", "record-3");
    }
}

2. 消费者：识别结束信号并停止容器 #

通过注入KafkaListenerEndpointRegistry来管理监听容器，在消息处理方法中判断是否收到结束标记，一旦识别到就停止对应的容器；如果需要关闭整个应用，还可以触发Spring上下文退出：

@Service
public class KafkaRecordConsumer {

    private final KafkaListenerEndpointRegistry containerRegistry;
    private final ApplicationContext applicationContext;

    public KafkaRecordConsumer(KafkaListenerEndpointRegistry containerRegistry,
                               ApplicationContext applicationContext) {
        this.containerRegistry = containerRegistry;
        this.applicationContext = applicationContext;
    }

    @KafkaListener(id = "business-consumer", topics = TARGET_TOPIC, concurrency = "2")
    public void consumeRecord(String message, Acknowledgment acknowledgment) {
        // 识别结束标记
        if ("END_OF_STREAM".equals(message)) {
            // 先确认该结束消息，避免重复消费
            acknowledgment.acknowledge();

            // 获取当前监听容器并停止
            MessageListenerContainer container = containerRegistry.getListenerContainer("business-consumer");
            if (container != null && container.isRunning()) {
                container.stop();
                System.out.println("Consumer container stopped");
            }

            // 如果需要彻底关闭整个应用，执行以下代码
            int exitCode = SpringApplication.exit(applicationContext, () -> 0);
            System.exit(exitCode);
            return;
        }

        // 处理业务消息逻辑
        processBusinessRecord(message);
        acknowledgment.acknowledge();
    }

    private void processBusinessRecord(String message) {
        System.out.println("Processing business record: " + message);
        // 替换为实际业务处理代码
    }
}

3. 关于KafkaListenerContainerFactory的说明 #

KafkaListenerContainerFactory的作用是创建MessageListenerContainer实例，它本身不需要主动停止——当所有由它创建的容器都停止后，工厂只是一个闲置的Bean。如果需要彻底销毁工厂，可以在Spring上下文关闭时自动销毁（上下文关闭时会销毁所有单例Bean），无需额外代码。

关键注意事项 #

• 并发消费场景：如果消费者设置了concurrency>1，需要发送与并发数相同数量的结束标记，确保每个消费者实例都能收到停止信号。
• 消息发送可靠性：使用send().get()同步等待发送确认，避免结束标记提前发送导致消费者遗漏业务消息。
• 消息确认：处理结束标记时一定要先调用acknowledge()，避免该消息被重新消费。

内容的提问来源于stack exchange，提问作者Rohit Verma