如何基于WebFlux以响应式方式通过Spring Cloud Stream向Kafka Topic发消息（不使用StreamBridge）

下面是一套完全基于响应式编程模型、不依赖StreamBridge的实现方案，适配Spring Cloud Stream的函数式编程规范：

1. 添加核心依赖 #

在你的构建文件中引入WebFlux和Spring Cloud Stream Kafka的依赖：

Maven（pom.xml） #

<dependencies>
    <!-- Spring WebFlux 响应式Web框架 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <!-- Spring Cloud Stream Kafka 绑定器 -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-kafka</artifactId>
    </dependency>
</dependencies>

Gradle（build.gradle） #

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
    implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka'
}

2. 配置响应式生产者绑定 #

通过Spring Cloud Stream的函数式模型，结合Reactor的Sinks创建响应式消息发送通道：

生产者配置类 #

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.function.Supplier;

@Configuration
public class KafkaProducerConfig {

    // 创建一个Sink用于接收待发送的消息
    @Bean
    public Sinks.Many<Message<String>> messageSink() {
        // unicast适用于单消费者场景，若需多消费可改用many().multicast()
        return Sinks.many().unicast().onBackpressureBuffer();
    }

    // 定义Spring Cloud Stream的Supplier函数，将Sink中的消息转发到Kafka
    @Bean
    public Supplier<Flux<Message<String>>> messageProducer(Sinks.Many<Message<String>> sink) {
        return sink::asFlux;
    }
}

3. 编写响应式生产者服务 #

封装消息发送逻辑，提供单条/批量发送的响应式方法：

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

@Service
public class KafkaReactiveProducerService {

    private final Sinks.Many<Message<String>> messageSink;

    public KafkaReactiveProducerService(Sinks.Many<Message<String>> messageSink) {
        this.messageSink = messageSink;
    }

    // 发送单条消息
    public Mono<Void> sendMessage(String payload) {
        Message<String> message = MessageBuilder.withPayload(payload).build();
        return Mono.fromRunnable(() -> 
            messageSink.emitNext(message, Sinks.EmitFailureHandler.FAIL_FAST)
        ).then();
    }

    // 批量发送消息
    public Mono<Void> sendMessages(Flux<String> payloads) {
        return payloads
                .map(payload -> MessageBuilder.withPayload(payload).build())
                .doOnNext(message -> 
                    messageSink.emitNext(message, Sinks.EmitFailureHandler.FAIL_FAST)
                )
                .then();
    }
}

4. 编写WebFlux控制器 #

暴露HTTP接口，接收前端请求并触发消息发送：

import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
public class MessageController {

    private final KafkaReactiveProducerService producerService;

    public MessageController(KafkaReactiveProducerService producerService) {
        this.producerService = producerService;
    }

    // 发送单条文本消息
    @PostMapping("/send")
    public Mono<Void> sendSingleMessage(@RequestBody String payload) {
        return producerService.sendMessage(payload);
    }

    // 批量发送消息（接收JSON数组）
    @PostMapping("/send-many")
    public Mono<Void> sendBatchMessages(@RequestBody Flux<String> payloads) {
        return producerService.sendMessages(payloads);
    }
}

5. 配置文件（application.yml） #

配置Kafka地址、Topic名称及绑定规则：

spring:
  cloud:
    stream:
      function:
        definition: messageProducer # 对应配置类中Supplier的bean名称
      bindings:
        # 函数式绑定的命名规则：函数名-out-索引（默认索引为0）
        messageProducer-out-0:
          destination: your-target-topic # 替换为你的Kafka Topic名称
      kafka:
        binder:
          brokers: localhost:9092 # 替换为你的Kafka Broker地址
          configuration:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: org.apache.kafka.common.serialization.StringSerializer

6. 验证方案 #

1. 启动Kafka服务并创建对应的Topic
2. 启动Spring Boot应用
3. 用curl发送测试请求：
   • 单条消息：

     curl -X POST -H "Content-Type: text/plain" -d "Hello Reactive Kafka" http://localhost:8080/send
     

   • 批量消息：

     curl -X POST -H "Content-Type: application/json" -d '["Msg 1", "Msg 2", "Msg 3"]' http://localhost:8080/send-many
     

4. 通过Kafka消费者工具验证消息是否成功投递

注意事项 #

• 若需要发送带自定义Headers的消息，可通过MessageBuilder.setHeader(key, value)添加
• Sink的类型可根据实际场景调整：multicast()适用于多消费者，unicast()适用于单消费者
• EmitFailureHandler的策略可按需修改，比如FAIL_FAST会在发送失败时立即抛出异常，也可自定义重试逻辑

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