Spring Cloud Function接口返回成功/失败处理及消息确认方案咨询

有两种可行的实现方式，分别对应保留Function接口写法和对齐原有逻辑两种场景： #

方案1：使用Spring Cloud Stream内置发送事件回调，保留Function写法 #

Spring Cloud Stream Kafka Binder会在消息发送完成后自动发布对应事件，你不需要自己管理KafkaTemplate，只需要监听事件即可实现发送结果感知：

步骤1：改造现有Function逻辑 #

不要在Function内部提前执行ack，把确认对象和业务标识放到返回消息的Header中传递：

@Bean
public Function<Message<NotificationMessage>, Message<ValidatedEvent>> validatedProducts() {
    return message -> {
        Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
        NotificationMessage notificationMessage = message.getPayload();
        // 存储初始消费记录
        notificationMessageService.saveOrUpdate(notificationMessage, 0, false);
        // 调用外部服务、处理数据
        String status = restEndpoint.getStatusFor(notificationMessage);
        ValidatedEvent event = getProcessingResult(notificationMessage, status);
        // 构造返回消息，传递ack对象和业务主键
        return MessageBuilder
                .withPayload(event)
                .setHeader(KafkaHeaders.MESSAGE_KEY, event.getKey().getBytes())
                .setHeader("custom_ack", acknowledgment)
                .setHeader("notification_id", notificationMessage.getId())
                .build();
    };
}

步骤2：添加发送事件监听器 #

分别监听发送成功和失败事件，执行对应的DB更新和消息确认逻辑：

@Component
public class KafkaSendResultHandler {
    @Autowired
    private NotificationMessageService notificationMessageService;

    // 处理发送成功逻辑
    @EventListener
    public void onSendSuccess(KafkaSendSuccessEvent event) {
        Message<?> outboundMsg = event.getMessage();
        Acknowledgment ack = outboundMsg.getHeaders().get("custom_ack", Acknowledgment.class);
        Long notificationId = outboundMsg.getHeaders().get("notification_id", Long.class);
        // 更新DB状态为成功
        notificationMessageService.saveOrUpdate(notificationId, 1, true);
        // 确认消费消息
        Optional.ofNullable(ack).ifPresent(Acknowledgment::acknowledge);
    }

    // 处理发送失败逻辑
    @EventListener
    public void onSendFailure(KafkaSendFailureEvent event) {
        Message<?> outboundMsg = event.getMessage();
        Long notificationId = outboundMsg.getHeaders().get("notification_id", Long.class);
        // 更新DB状态为失败，不执行ack，框架会按配置走重试/死信逻辑
        notificationMessageService.saveOrUpdate(notificationId, 1, false);
    }
}

适用场景 #

希望复用Spring Cloud Function的自动流转能力，不需要手动管理消息发送的场景，Spring Cloud Stream 3.x及以上版本均默认支持这两个事件，无需额外配置。 #

方案2：改用Consumer接口，对齐原有实现逻辑 #

如果不想适配框架的自动发送逻辑，直接把Function改为Consumer，手动调用KafkaTemplate发送消息，逻辑和原有@StreamListener实现完全一致，改动最小：

@Bean
public Consumer<Message<NotificationMessage>> validatedProducts() {
    return message -> {
        Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
        NotificationMessage notificationMessage = message.getPayload();
        try {
            notificationMessageService.saveOrUpdate(notificationMessage, 0, false);
            String status = restEndpoint.getStatusFor(notificationMessage);
            ValidatedEvent event = getProcessingResult(notificationMessage, status);
            
            Message<ValidatedEvent> outMessage = MessageBuilder
                    .withPayload(event)
                    .setHeader(KafkaHeaders.MESSAGE_KEY, event.getKey().getBytes())
                    .build();
            // 同步等待发送结果，也可使用addCallback实现异步回调
            kafkaTemplate.send(outMessage).get();
            
            notificationMessageService.saveOrUpdate(notificationMessage, 1, true);
            Optional.ofNullable(acknowledgment).ifPresent(Acknowledgment::acknowledge);
        } catch (Exception e) {
            notificationMessageService.saveOrUpdate(notificationMessage, 1, false);
        }
    };
}

适用场景 #

希望保留原有逻辑的流程控制，降低迁移适配成本的场景，排查问题也更直接。

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