Spring Cloud Stream Kafka Binder函数式编程无法收发消息排查

一、核心问题分析 #

你遇到的两个问题本质上都是对Spring Cloud Stream 3.x函数式编程模型的配置逻辑不熟悉导致的，以下是具体原因和解决办法：

1. 生产者Bean创建成功但无消息发送 #

Supplier<Integer>类型的生产者默认是懒加载、按需执行的，不会自动触发消息发送，这是3.x函数式模型的设计逻辑。常见触发方式有两种：

方式一：用Flux实现持续消息生产

如果需要周期性发送消息，将Supplier返回类型改为Flux<Integer>，通过响应式流持续输出数据：

@Bean
public Supplier<Flux<Integer>> numberProducer() {
    return () -> Flux.interval(Duration.ofSeconds(1))
                     .map(Long::intValue)
                     .log("Producer-Log");
}

方式二：配置轮询器触发Supplier

如果是普通Supplier，通过配置轮询规则让框架定时调用：

spring:
  cloud:
    stream:
      poller:
        fixed-delay: 1000 # 每隔1秒调用一次Supplier

绑定配置检查

必须确保application.yml中的绑定名称和Bean名称对应，格式为<beanName>-out-0：

spring:
  cloud:
    stream:
      kafka:
        binder:
          bootstrap-servers: localhost:9092
      bindings:
        numberProducer-out-0: # 对应上面的Supplier Bean名称numberProducer
          destination: test-topic
          content-type: application/json

2. 不加@EnableBinding找不到StreamBridge #

@EnableBinding是Spring Cloud Stream 2.x的旧版绑定注解，3.x函数式模型完全不需要这个注解。找不到StreamBridge的原因通常是：

原因一：依赖或版本不兼容

• 必须引入正确的Kafka binder依赖，不要混用旧版spring-cloud-stream依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    <version>3.2.7</version>
</dependency>

• 确保Spring Boot版本与Stream版本匹配：3.2.7对应的Spring Boot版本是2.6.x（如2.6.13），版本不匹配会导致自动配置失效。

原因二：函数式模式未正确启用

默认情况下3.x已开启函数式模式，但如果手动配置了spring.cloud.stream.function.definition，必须确保包含你定义的函数Bean名称，否则自动配置的组件（包括StreamBridge）可能无法正常加载：

spring:
  cloud:
    stream:
      function:
        definition: numberProducer;numberConsumer # 多个函数用分号分隔

二、正确的消费者配置示例 #

确保消费者Bean的绑定配置格式为<beanName>-in-0：

@Bean
public Consumer<Integer> numberConsumer() {
    return num -> System.out.println("Received message: " + num);
}

spring:
  cloud:
    stream:
      bindings:
        numberConsumer-in-0:
          destination: test-topic
          content-type: application/json

三、手动发送消息推荐方案 #

如果需要手动触发消息发送（而非持续生产），建议直接使用StreamBridge，不需要定义Supplier Bean：

@RestController
public class MessageSender {
    private final StreamBridge streamBridge;

    public MessageSender(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    @GetMapping("/send")
    public String send() {
        streamBridge.send("test-topic", 12345);
        return "Message sent";
    }
}

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