Spring Cloud Stream是否支持ReactiveKafkaProducerTemplate绑定及替代Sink方案？

为什么可以替代？ #

Reactive Kafka Binder 1.x完全兼容Spring Cloud Stream的响应式编程模型，既可以沿用你当前的函数式Supplier绑定模式，也能直接通过ReactiveKafkaProducerTemplate发送消息，完全覆盖现有基于Sinks的生产者逻辑。

为什么应当替代？ #

• 原生集成Spring Cloud Stream的Kafka配置体系，无需自行实现Sink到Kafka的桥接逻辑，大幅减少自定义代码量
• ReactiveKafkaProducerTemplate封装了Kafka生产者的响应式操作，自动处理生产者生命周期、线程池管理、消息重试等细节，比自定义Sinks方案更稳定可靠
• 天然支持Kafka高级特性（事务、分区策略、消息键设置等），这些特性在自定义Sinks中需要大量额外代码才能实现
• 贴合Spring生态的响应式编程模型，后续维护和扩展成本更低

Kotlin用法示例 #

1. 核心依赖（build.gradle.kts）

dependencies {
    implementation("org.springframework.cloud:spring-cloud-stream-binder-kafka-reactive")
    implementation("org.springframework.kafka:spring-kafka-reactive")
}

2. 应用配置（application.yml）

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      bindings:
        projekte-out-0:
          destination: projekte-topic # 目标Kafka主题
          content-type: application/json
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

3. 生产者实现

@Component
class GenericEventProducer {
    // 注入ReactiveKafkaProducerTemplate，用于直接发送消息
    @Autowired
    private lateinit var producerTemplate: ReactiveKafkaProducerTemplate<String, BaseEvent<*>>

    // 保留原方案的Supplier模式，通过Spring Cloud Stream自动绑定到Kafka
    @Bean
    fun projekte(): Supplier<Flux<Message<ProjektEvent>>> {
        // 用Sink收集事件（兼容原有事件收集逻辑）
        val eventSink = Sinks.many().replay().latest<ProjektEvent>()
        return Supplier {
            eventSink.asFlux()
                .map { 
                    MessageBuilder.withPayload(it)
                        .setHeader(KafkaHeaders.TOPIC, "projekte-topic")
                        .build() 
                }
        }
    }

    // 替代原sink.tryEmitNext的发送方法
    fun sendEvent(event: BaseEvent<*>) {
        producerTemplate.send("projekte-topic", event)
            .doOnSuccess { result -> println("事件发送成功，offset: ${result.recordMetadata().offset()}") }
            .doOnError { ex -> println("事件发送失败：${ex.message}") }
            .subscribe() // 非阻塞订阅触发发送
    }
}

注意事项 #

• 若保留Supplier模式，Spring Cloud Stream会自动将Flux中的消息发送到配置的projekte-out-0绑定对应的主题
• 直接使用producerTemplate时，可手动指定主题，适合动态选择主题的场景
• 可通过MessageHeader设置Kafka分区、消息键等属性，比如setHeader(KafkaHeaders.MESSAGE_KEY, event.id)

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