如何使用Spring Cloud Kafka Stream 3.1函数式编程定义生产者

关于废弃的@Input、@Output注解 #

从Spring Cloud Stream 3.x版本开始，官方全面切换到函数式编程模型作为唯一推荐的实现方案，@Input、@Output属于2.x时代命令式编程的专属注解：

• 虽部分3.x版本还保留兼容逻辑可运行，但已标注废弃，在3.1版本的Kafka Stream场景中使用会触发Binder兼容层的额外Bean注册逻辑，容易引发通道绑定冲突
• 纯生产者场景完全不需要使用这类输入相关的注解，没有引入必要

纯生产者无单独案例的原因 #

你查阅的Kafka Stream Binder文档，核心定位是流处理链路的实现说明，默认围绕「消费-处理-生产」的完整流场景编写示例，纯生产者属于无输入源的单独生产场景，通用实现逻辑放在Spring Cloud Stream核心的通用特性章节，不属于Kafka Binder专属能力，因此在Binder专属文档中不会单独列示。

根据触发生产消息的场景不同，有两种成熟实现方案：

方案1：业务事件触发型（最常用） #

适合业务代码执行到特定节点（如用户下单、数据更新）时主动发消息的场景，使用官方提供的StreamBridge工具类实现，不需要定义任何函数Bean：

1. 引入依赖

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

2. 核心配置（application.yml）

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: 你的Kafka集群地址
          defaultBrokerPort: 9092
      bindings:
        # 配置目标主题的生产者属性
        your-target-topic:
          content-type: application/json
          producer:
            configuration:
              key.serializer: org.apache.kafka.common.serialization.StringSerializer
              value.serializer: org.springframework.kafka.support.serializer.JsonSerializer

3. 业务代码中直接调用发送

import lombok.RequiredArgsConstructor;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class OrderService {
    // 直接注入StreamBridge即可
    private final StreamBridge streamBridge;

    public void createOrder(OrderDTO order) {
        // 业务处理逻辑...
        // 第一个参数为目标主题名，第二个为消息内容
        streamBridge.send("your-target-topic", order);
    }
}

方案2：定时触发型 #

适合需要周期性轮询数据、自动发送消息的场景，通过定义Supplier函数Bean实现，Binder会自动绑定输出通道：

1. 定义函数Bean

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.support.MessageBuilder;
import java.util.function.Supplier;

@Configuration
public class TriggerProducerConfig {
    // 函数命名规则：绑定通道默认生成[函数名]-out-0
    @Bean
    public Supplier<Message<OrderDTO>> orderPeriodicProducer() {
        return () -> {
            // 此处可编写查询待发送数据的逻辑，返回的内容会自动发送到绑定的主题
            OrderDTO pendingOrder = orderMapper.selectPendingSend();
            return MessageBuilder.withPayload(pendingOrder).build();
        };
    }
}

2. 通道绑定配置（application.yml）

spring:
  cloud:
    stream:
      bindings:
        # 绑定规则和函数名对应
        orderPeriodicProducer-out-0:
          destination: your-target-topic
          content-type: application/json
      poller:
        # 配置定时触发周期，单位毫秒
        fixed-delay: 5000

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