Spring Cloud Gateway无Controller场景下Kafka消息发送示例咨询

当然有可行的整合方案，核心思路是借助Spring Cloud Gateway的过滤器机制（全局过滤器或路由级过滤器），在请求转发的生命周期中嵌入Kafka消息发送逻辑，完全不需要额外编写Controller。以下是具体实现示例：

1. 添加核心依赖 #

在pom.xml（Maven）中引入Gateway和Kafka的starter：

<dependencies>
    <!-- Spring Cloud Gateway -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-gateway</artifactId>
    </dependency>
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
</dependencies>

2. 配置Kafka基础参数 #

在application.yml中配置Kafka生产者信息：

spring:
  kafka:
    bootstrap-servers: localhost:9092 # 替换为你的Kafka集群地址
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      retries: 3

3. 封装Kafka消息发送服务 #

创建一个通用的消息发送服务，封装KafkaTemplate的调用：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaMessageProducer {
    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaMessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

4. 自定义Gateway全局过滤器实现消息发送 #

编写一个GlobalFilter，在请求转发完成后（或根据需求在转发前）触发消息发送逻辑，比如将请求路径、响应状态等信息发送到Kafka：

import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

@Component
public class KafkaMessageFilter implements GlobalFilter, Ordered {
    private final KafkaMessageProducer kafkaProducer;
    private static final String KAFKA_TOPIC = "gateway-request-topic"; // 替换为你的目标Topic

    public KafkaMessageFilter(KafkaMessageProducer kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        // 先执行请求转发逻辑
        return chain.filter(exchange)
                .doFinally(signalType -> {
                    ServerHttpResponse response = exchange.getResponse();
                    // 构造消息内容，可根据业务需求定制
                    String message = String.format("请求路径: %s, 响应状态码: %d",
                            request.getPath(), response.getStatusCode().value());
                    // 发送消息到Kafka
                    kafkaProducer.sendMessage(KAFKA_TOPIC, message);
                });
    }

    @Override
    public int getOrder() {
        return Ordered.LOWEST_PRECEDENCE; // 确保过滤器在请求转发完成后执行
    }
}

5. 配置Gateway路由 #

在application.yml中配置你的转发路由，无需额外Controller：

spring:
  cloud:
    gateway:
      routes:
        - id: service-route
          uri: http://your-target-microservice:8080 # 替换为目标微服务地址
          predicates:
            - Path=/api/** # 匹配的请求路径

关键注意点 #

• 过滤器执行顺序：通过getOrder()方法调整，LOWEST_PRECEDENCE确保在转发完成后执行；若需要在转发前发送消息，可设置更高优先级（比如Ordered.HIGHEST_PRECEDENCE）。
• 消息内容定制：可根据业务需求，从ServerWebExchange中获取请求参数、请求体、响应体等信息，序列化后发送到Kafka。
• 异常处理：建议在sendMessage方法中添加异常捕获，避免Kafka发送失败影响网关请求转发流程。

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