Spring Boot中如何处理Apache Kafka队列消费完成后的操作

针对你的需求，有几种不同场景的实现方案，可根据业务实际情况选择：

方案1：改用批量消费模式（最简洁） #

如果业务允许批量拉取消息，直接使用@KafkaListener的批量消费能力，在批量处理完成后直接操作List即可。

示例代码：

@Component
public class KafkaConsumer {

    private List<Object> processedData = new ArrayList<>();

    // 指定批量消费的容器工厂
    @KafkaListener(topics = "your-topic", containerFactory = "batchListenerContainerFactory")
    public void consumeBatch(List<String> messages, Acknowledgment ack) {
        // 逐条处理消息并加入List
        messages.forEach(msg -> processedData.add(processMessage(msg)));
        
        // 所有批量消息处理完成后，执行List的后续操作
        postProcessList();
        
        // 手动提交偏移量（按需配置）
        ack.acknowledge();
    }

    private Object processMessage(String msg) {
        // 你的消息处理逻辑
        return msg.toUpperCase();
    }

    private void postProcessList() {
        System.out.println("批量消息处理完成，List大小：" + processedData.size());
        // 执行持久化、分析等业务操作
        // 操作完成后可清空List，准备下一批处理
        processedData.clear();
    }
}

配置说明：需要在容器工厂中开启批量消费，设置batchListener = true，并调整max.poll.records参数控制单次拉取的消息数量。

方案2：监听消费者空闲事件（逐条消费场景） #

如果必须保持逐条消费的模式，可以通过监听Spring Kafka发布的ConsumerIdleEvent，判断队列无消息可消费时触发List操作。

示例代码：

@Component
public class KafkaConsumer implements ApplicationListener<ConsumerIdleEvent> {

    // 线程安全的List，避免并发问题
    private final List<Object> processedData = Collections.synchronizedList(new ArrayList<>());

    // 指定消费者容器ID，用于后续事件匹配
    @KafkaListener(topics = "your-topic", id = "your-consumer-container")
    public void consumeSingle(String message, Acknowledgment ack) {
        processedData.add(processMessage(message));
        ack.acknowledge();
    }

    private Object processMessage(String msg) {
        // 你的消息处理逻辑
        return msg.toUpperCase();
    }

    @Override
    public void onApplicationEvent(ConsumerIdleEvent event) {
        // 确认是目标消费者触发的空闲事件
        if ("your-consumer-container".equals(event.getListenerId())) {
            postProcessList();
        }
    }

    private void postProcessList() {
        if (!processedData.isEmpty()) {
            System.out.println("队列消息已全部消费完成，List大小：" + processedData.size());
            // 执行后续业务操作
            processedData.clear();
        }
    }
}

配置说明：需要设置消费者参数idle.event.interval.ms（比如3000ms），指定多长时间无消息触发空闲事件。

方案3：基于偏移量判断消费完成（精准控制） #

如果需要精准判断某个分区的所有消息已消费，可以通过对比消费者已提交偏移量和分区末端偏移量来触发操作。

示例代码：

@Component
public class KafkaConsumer {

    private final List<Object> processedData = Collections.synchronizedList(new ArrayList<>());
    @Autowired
    private KafkaConsumer<String, String> kafkaConsumer;

    @KafkaListener(topics = "your-topic")
    public void consume(String message, ConsumerRecordMetadata metadata, Acknowledgment ack) {
        processedData.add(processMessage(message));
        ack.acknowledge();
        
        // 检查当前分区是否已消费完成
        checkPartitionConsumptionStatus(metadata);
    }

    private void checkPartitionConsumptionStatus(ConsumerRecordMetadata metadata) {
        TopicPartition topicPartition = new TopicPartition(metadata.topic(), metadata.partition());
        // 获取已提交的偏移量
        OffsetAndMetadata committedOffset = kafkaConsumer.committed(topicPartition);
        // 获取分区的最新末端偏移量
        long endOffset = kafkaConsumer.endOffsets(Collections.singleton(topicPartition)).get(topicPartition);
        
        // 已提交偏移量等于末端偏移量-1，说明分区所有消息已消费
        if (committedOffset != null && committedOffset.offset() == endOffset - 1) {
            postProcessList();
        }
    }

    // processMessage和postProcessList方法同前...
}

注意：KafkaConsumer不是线程安全的，多线程场景下需谨慎使用此方案；若topic有多个分区，需汇总所有分区的偏移量状态判断整体消费完成。

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