Spring Cloud API Gateway能否向RabbitMQ发消息？线程占用过高问题咨询

能否通过Spring API Gateway向RabbitMQ发送消息？ #

可以，但绝不能用你当前这种同步阻塞等待响应的方式——这完全违背了Gateway响应式架构的设计初衷，也是线程耗尽、应用崩溃的核心原因。

核心问题分析 #

你现在的实现是在Gateway内部通过Rest API路由请求后，同步发送RabbitMQ消息并等待响应，这种操作会直接阻塞Gateway的事件循环线程（Reactor的NIO线程）。Gateway的事件循环线程数量是固定的（默认和CPU核心数挂钩），一旦大量请求涌入，所有线程都被阻塞在等待MQ响应的环节，直接导致无法处理新请求，最终触发应用崩溃。

正确实现方案 #

必须遵循响应式编程的非阻塞原则，使用Spring AMQP的响应式客户端或结合Reactor的异步处理能力：

1. 使用响应式RabbitMQ客户端 #

先引入依赖：

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbitmq-stream</artifactId>
</dependency>

然后在Gateway的自定义过滤器中，以非阻塞方式发送消息并处理响应：

@Component
public class RabbitMqFilter implements GatewayFilter {

    private final StreamSender sender;
    private final StreamReceiver receiver;

    public RabbitMqFilter(StreamSender sender, StreamReceiver receiver) {
        this.sender = sender;
        this.receiver = receiver;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 从请求体中提取消息内容
        return exchange.getRequest().getBody()
                .map(DataBuffer::asInputStream)
                .map(inputStream -> {
                    try (InputStream is = inputStream) {
                        return new String(is.readAllBytes(), StandardCharsets.UTF_8);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                })
                .flatMap(message -> {
                    // 生成唯一关联ID，用于匹配响应
                    String correlationId = UUID.randomUUID().toString();
                    Message streamMessage = MessageBuilder.withBody(message.getBytes())
                            .setHeader("correlationId", correlationId)
                            .setHeader("replyTo", "response-queue")
                            .build();

                    // 异步发送消息+订阅响应队列
                    return sender.send(streamMessage)
                            .then(receiver.receive("response-queue")
                                    .filter(response -> correlationId.equals(response.getMessage().getHeaders().get("correlationId")))
                                    .next()
                                    .timeout(Duration.ofSeconds(10)) // 添加超时控制，避免请求挂起
                                    .map(response -> {
                                        // 将MQ响应写入Gateway的响应对象
                                        exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
                                        DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(response.getMessage().getBody());
                                        return exchange.getResponse().writeWith(Mono.just(buffer));
                                    })
                                    .flatMap(mono -> mono));
                });
    }
}

2. 配置路由使用自定义过滤器 #

替换原来的URI路由规则，让目标路径走自定义过滤器：

@Configuration
public class GatewayConfig {

    @Bean
    public RouteLocator customRouteLocator(RouteLocatorBuilder builder, RabbitMqFilter rabbitMqFilter) {
        return builder.routes()
                .route("mq-route", r -> r.path("/mq/**")
                        .filters(f -> f.filter(rabbitMqFilter))
                        .uri("no://op")) // 用占位URI，实际逻辑由过滤器处理
                .build();
    }
}

3. 关键优化点 #

• 禁止同步阻塞：绝对不能调用block()这类会阻塞线程的方法，所有MQ交互必须异步非阻塞
• 复用事件循环线程：响应式客户端会复用Gateway的NIO线程，不会额外创建阻塞线程
• 强制超时控制：必须给MQ发送和响应添加超时，防止请求无限期挂起占用资源

替代方案：引入中间层服务 #

如果不想修改Gateway代码，可以在Gateway和RabbitMQ之间加一个轻量的Spring WebFlux服务，让Gateway只负责路由请求到这个中间服务，由中间服务处理RabbitMQ的消息发送和响应逻辑。这样Gateway仅做转发，不会占用线程处理MQ操作。

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