Spring Cloud Stream Kafka生产者强制flush及异常捕获咨询

缓存消息强制刷出实现方案 #

你当前使用的是Spring Integration Kafka的消息通道编程模型，要避免生产者内存缓存消息因进程崩溃丢失，可通过以下两种等价方式实现：

• 方案1：在批次最后一条消息上设置KafkaIntegrationHeaders.FLUSH值为true。消息处理器处理完该消息后，会自动调用底层KafkaProducer.flush()方法，阻塞等待当前生产者缓冲区中所有未发送的消息完成网络传输、拿到Broker端的ACK确认后，才会从output.send()方法返回，效果和直接调用KafkaTemplate.flush()完全一致。
• 方案2：直接注入对应生产者的KafkaTemplate实例，在批次消息全部发送完成后手动调用kafkaTemplate.flush()方法，逻辑和设置FLUSH头完全相同。

注意：仅调用flush不能100%保证消息不丢，需要同步配置生产者参数：acks=all、enable.idempotence=true、合理配置重试次数，才能保证消息被Broker端所有ISR副本持久化，不会因Broker宕机丢失。

FLUSH模式下的异常捕获规则 #

你代码中编写的try-catch块可以捕获到发送流程中抛出的同步异常，具体规则如下：

• 触发FLUSH后，output.send()方法会变为阻塞调用，等待所有待发送消息的发送Future完成。如果发送过程中出现不可恢复的异常（包括认证失败、IO异常且重试耗尽、序列化错误、Broker端拒绝写入、事务提交失败等），异常会被包装为MessageHandlingException直接在当前调用线程抛出，完全可以被方法内的catch块捕获。
• 如果不设置FLUSH头，普通output.send()是异步非阻塞调用，发送异常只会触发生产者回调或全局错误监听器，不会在send()调用点抛出，无法被当前方法的catch块捕获。

代码修正提示 #

Spring Messaging的Message对象是不可变类，不存在setHeader方法，直接调用会抛出UnsupportedOperationException，正确的写法参考如下：

public void process(List<String> input) {
    // 前置业务逻辑
    try {
        Message<?> message = MessageBuilder.withPayload(/* 填入实际消息负载 */)
                .setHeader(KafkaIntegrationHeaders.FLUSH, true)
                .build();
        output.send(message);
        // 代码执行到此处说明所有缓存消息已完成Broker确认，无发送异常
    } catch (Exception e) {
        // 此处可捕获flush等待过程中抛出的所有发送异常
        e.printStackTrace();
        // 可在此处实现重试、降级、告警等容错逻辑
    }
}

补充：如果当前消息通道配置了全局错误通道，异常会优先路由到错误通道，未被错误通道消费的异常才会向上抛给调用点的catch块。

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