Spring Cloud Stream 2.0升级3.1：Kinesis消息发送改造疑问

原2.0版本实现（基于@Output注解） #

业务服务类 #

@Component
public class KinesisStreamService {
    private final Log logger = LogFactory.getLog(this.getClass());
    private StreamProcessor orderOut;

    @Autowired
    public KinesisStreamService(StreamProcessor orderOut) {
        this.orderOut = orderOut;
    }

    public void send(String event) {
        if (event != null) {
            this.orderOut.ordersOut().send(new GenericMessage(event));
            this.logger.debug("Event sent KinesisInStreamServiceImpl : " + event);
        } else {
            throw new RuntimeException("Event can not be null");
        }
    }
}

通道定义接口 #

@Service
public interface StreamProcessor {
    String INPUT = "ordersIn";

    @Output
    MessageChannel ordersOut();

    @Input
    SubscribableChannel ordersIn();
}

升级中的困惑与问题 #

在3.1版本中，@Output注解已被废弃，官方推荐使用函数式编程模型，示例中是创建Supplier Bean：

@Service
class PubSubSendQueue {
    @Bean
    public Supplier<String> output(){
        return () -> "Adam";
    }
}

但实际业务中需要在运行时传入参数动态发送消息，尝试了以下带@Lazy的写法，不确定是否可行：

@Bean
@Lazy
public Supplier<GenericMessage> ordersOut(String event) {
    return () -> new GenericMessage(event);
}

当前项目依赖（Gradle） #

dependencies {
    implementation 'org.apache.activemq:activemq-broker:5.17.2'
    implementation 'com.amazonaws:aws-java-sdk:1.12.310'
    implementation 'com.amazonaws:amazon-sqs-java-messaging-lib:1.1.0'
    implementation 'org.springframework.integration:spring-integration-aws:2.5.4'
    implementation 'org.springframework.cloud:spring-cloud-stream-binder-kinesis:3.0.0'
    implementation 'org.springframework:spring-web:6.0.0'
    implementation 'org.springframework.boot:spring-boot-starter:3.0.5'
    implementation 'javax.xml.bind:jaxb-api:2.3.0'
    implementation 'com.sun.xml.bind:jaxb-impl:2.3.6'
    implementation 'org.glassfish.jaxb:jaxb-runtime:2.3.6'
    implementation 'javax.activation:activation:1.1.1'
    implementation 'org.apache.httpcomponents.client5:httpclient5:5.2.1'
    testImplementation 'org.codehaus.groovy:groovy-all:2.5.18'
    testImplementation 'org.spockframework:spock-spring:1.3-groovy-2.5'
    testImplementation 'org.spockframework:spock-core:1.3-groovy-2.5'
    testImplementation 'org.springframework:spring-test:6.0.0'
    implementation 'cglib:cglib-nodep:3.2.7'
    implementation group: 'org.springframework', name: 'spring-jms', version: '6.0.0'
    implementation 'com.fasterxml.jackson.core:jackson-databind:2.14.0-rc1'
    implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-cbor', version: '2.13.4'
    implementation 'javax.jms:javax.jms-api:2.0.1'
    implementation 'javax.validation:validation-api:2.0.1.Final'
    implementation "jakarta.xml.bind:jakarta.xml.bind-api:2.3.2"
    implementation "org.glassfish.jaxb:jaxb-runtime:2.3.2"
}

解决方案 #

核心结论：不要用Supplier实现按需发送 #

Spring Cloud Stream的Supplier函数式模型专为主动触发的消息生成场景设计（比如定时轮询、周期推送），完全不适合替代原有的MessageChannel.send()这种业务触发的按需发送方式。要实现运行时传参动态发送，正确的工具是StreamBridge。

具体实现 #

1. 修改业务服务类，注入StreamBridge

@Component
public class KinesisStreamService {
    private final Log logger = LogFactory.getLog(this.getClass());
    private final StreamBridge streamBridge;

    @Autowired
    public KinesisStreamService(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    public void send(String event) {
        if (event != null) {
            // "ordersOut"对应原@Output绑定的通道名称，需在配置中对应设置目标Kinesis流
            streamBridge.send("ordersOut", new GenericMessage(event));
            this.logger.debug("Event sent KinesisInStreamServiceImpl : " + event);
        } else {
            throw new RuntimeException("Event can not be null");
        }
    }
}

2. 补充配置文件（以YAML为例）

spring:
  cloud:
    stream:
      bindings:
        ordersOut:
          destination: your-kinesis-stream-name # 替换为你的Kinesis流名称
          binder: kinesis

关于@Lazy Supplier方案的问题 #

你尝试的写法存在根本问题：

• Spring容器初始化Bean时无法自动注入运行时的event参数，会直接抛出依赖缺失异常
• @Lazy仅延迟Bean初始化，无法解决运行时动态传参的需求
• Supplier的设计逻辑是无参数的消息生成器，不支持接收外部业务参数

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