Spring Cloud Stream函数式处理：消息发布后如何收回应用控制权？

存在对应的配置选项，能够确保消息发布完成并收到Broker确认后，再将控制权交还给应用，从而安全执行数据库状态更新这类后续逻辑，具体方案如下：

1. 同步发送确认配置 #

通过配置生产者的同步发送属性，让消息发送操作阻塞直至收到Broker的确认回执，之后才会继续执行应用后续代码。

针对你提供的toUpperCase函数式绑定，默认输出绑定名称为toUpperCase-out-0，对应的配置如下：

# 开启同步发送，等待Broker确认
spring.cloud.stream.bindings.toUpperCase-out-0.producer.sync=true
# 可选：设置发送超时时间，避免无限阻塞
spring.cloud.stream.bindings.toUpperCase-out-0.producer.send-timeout=5000

2. 函数式场景的代码逻辑配合 #

当开启sync=true后，函数的返回值处理流程会等待消息成功发送并得到Broker确认，之后才会完成函数的执行周期。此时你可以在函数的后续逻辑（或者调用该函数的业务代码中）安全执行数据库状态更新操作，无需担心消息未完成发布就执行数据库操作的问题。

示例代码调整如下：

@Bean
public Function<String, String> toUpperCase() {
    return s -> {
        String upperCaseStr = s.toUpperCase();
        // sync=true会确保消息发送确认后才完成函数执行
        return upperCaseStr;
    };
}

@Service
public class BusinessService {
    private final JdbcTemplate jdbcTemplate;
    private final FunctionCatalog functionCatalog;

    public BusinessService(JdbcTemplate jdbcTemplate, FunctionCatalog functionCatalog) {
        this.jdbcTemplate = jdbcTemplate;
        this.functionCatalog = functionCatalog;
    }

    public void processMessage(String message) {
        // 执行函数式消息转换与发布，等待确认完成
        functionCatalog.lookup("toUpperCase").apply(message);
        // 消息发布确认后，执行数据库更新
        jdbcTemplate.update("UPDATE business_state SET status = ? WHERE message_id = ?", "COMPLETED", extractMessageId(message));
    }

    private String extractMessageId(String message) {
        // 自定义消息ID提取逻辑
        return message.split(":")[0];
    }
}

3. 可选：事务级别的一致性保障 #

如果需要消息发布与数据库更新操作具备事务一致性（要么都成功，要么都回滚），可以开启生产者事务配置：

spring.cloud.stream.bindings.toUpperCase-out-0.producer.transactional=true
spring.cloud.stream.producer.transaction-id-prefix=tx-

此时需配合Spring事务管理器，将消息发送和数据库更新放在同一个事务上下文里，进一步保障数据一致性。

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