WebFlux调用SCS应用：如何用函数式风格创建无过时注解的请求响应MessagingGateway

针对你想用WebFlux Boot调用独立SCS Boot应用，且不使用@Input、@Output、@EnableBinding这些废弃注解的需求，以下是基于Spring Cloud Stream最新函数式模型的实现方案：

一、SCS Boot 服务端（处理请求并返回响应） #

服务端通过函数式Function组件实现请求处理，自动绑定消息通道：

1. 核心代码 #

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.function.Function;

@Configuration
public class StreamProcessorConfig {

    // 定义请求处理函数：接收请求消息，返回响应消息
    @Bean
    public Function<String, String> requestHandler() {
        return request -> {
            // 这里替换为你的业务逻辑
            return "已处理请求：" + request;
        };
    }
}

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

指定函数绑定的消息主题/队列：

spring:
  cloud:
    stream:
      bindings:
        requestHandler-in-0:
          destination: request-channel # 请求消息通道
        requestHandler-out-0:
          destination: reply-channel # 响应消息通道
      function:
        definition: requestHandler # 启用该函数

二、WebFlux Boot 客户端（发送请求+接收响应） #

客户端用StreamBridge发送请求，结合函数式Consumer监听响应，同时用请求ID关联请求与响应，适配WebFlux异步特性：

1. 核心代码 #

import org.springframework.cloud.stream.function.StreamBridge;
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.Mono;
import reactor.core.publisher.MonoSink;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;

@Configuration
public class RequestReplyClient {

    private final ConcurrentMap<String, MonoSink<String>> requestMap = new ConcurrentHashMap<>();
    private final StreamBridge streamBridge;

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

    // 对外暴露的请求发送方法，返回Mono适配WebFlux
    public Mono<String> sendRequest(String requestContent) {
        String requestId = UUID.randomUUID().toString();
        Message<String> requestMsg = MessageBuilder.withPayload(requestContent)
                .setHeader("request-id", requestId)
                .build();

        return Mono.create(sink -> {
            requestMap.put(requestId, sink);
            streamBridge.send("request-channel", requestMsg);
        });
    }

    // 监听响应通道的函数，自动绑定到配置的响应通道
    @Bean
    public Consumer<Message<String>> replyListener() {
        return replyMsg -> {
            String requestId = replyMsg.getHeaders().get("request-id", String.class);
            MonoSink<String> sink = requestMap.remove(requestId);
            if (sink != null) {
                sink.success(replyMsg.getPayload());
            }
        };
    }
}

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

spring:
  cloud:
    stream:
      bindings:
        replyListener-in-0:
          destination: reply-channel # 监听响应通道
      function:
        definition: replyListener # 启用响应监听函数

3. WebFlux控制器调用示例 #

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.Mono;

@RestController
public class RequestController {

    private final RequestReplyClient requestClient;

    public RequestController(RequestReplyClient requestClient) {
        this.requestClient = requestClient;
    }

    @PostMapping("/send")
    public Mono<String> send(@RequestBody String request) {
        return requestClient.sendRequest(request);
    }
}

关键说明 #

• 全程未使用任何废弃注解，完全基于Spring Cloud Stream函数式编程模型构建。
• 通过请求ID关联请求与响应，解决异步场景下的消息对应问题，完美适配WebFlux的非阻塞特性。
• 服务端的Function自动完成输入输出通道绑定，客户端用StreamBridge灵活发送消息，Consumer监听响应，符合最新的Spring生态设计风格。

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