Spring Cloud Stream中Function<>如何实现消息可选生产？

一、Function<> 实现选择性生产消息 #

不需要重构为Consumer，直接在已有的Function中通过控制返回值即可实现按需发送：

• 当不需要发送消息时，返回null或用Optional包装返回Optional.empty()，Spring Cloud Stream 3.x及以上版本会自动跳过消息发送逻辑
• 保留原有的in/out绑定配置，无需额外修改

示例代码：

@Bean
public Function<String, Optional<String>> process() {
    return input -> {
        // 自定义判断逻辑，仅满足条件时返回消息
        if (input != null && input.startsWith("send:")) {
            return Optional.of(input.replace("send:", ""));
        }
        // 不满足条件时返回空，不会触发out绑定的消息发送
        return Optional.empty();
    };
}

二、Consumer<> + StreamBridge 循环引用问题解决 #

循环引用通常是Bean初始化顺序导致的，可通过以下方式解决：

1. 构造注入StreamBridge（推荐） #

通过构造方法注入StreamBridge，明确Spring的Bean依赖顺序，避免循环依赖：

@Component
public class MessageConsumer {
    private final StreamBridge streamBridge;

    // 构造方法注入，让Spring优先初始化StreamBridge
    public MessageConsumer(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    @Bean
    public Consumer<String> consume() {
        return input -> {
            if (input.contains("trigger")) {
                // 按需发送到指定的out绑定
                streamBridge.send("out-binding", input);
            }
        };
    }
}

2. 延迟初始化StreamBridge #

如果使用字段注入，添加@Lazy注解让StreamBridge在首次使用时才初始化：

@Component
public class MessageConsumer {
    @Lazy
    @Autowired
    private StreamBridge streamBridge;

    @Bean
    public Consumer<String> consume() {
        return input -> {
            if (input.contains("trigger")) {
                streamBridge.send("out-binding", input);
            }
        };
    }
}

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