Spring+Ribbon环境下如何拦截请求用于数据库回滚重放？

@Component
public class FeignDataSyncInterceptor implements feign.RequestInterceptor {

    private static final Logger log = LoggerFactory.getLogger(FeignDataSyncInterceptor.class);
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    public FeignDataSyncInterceptor(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }

    @Override
    public void apply(RequestTemplate template) {
        String method = template.method();
        if (Set.of("POST", "PUT", "DELETE").contains(method.toUpperCase())) {
            try {
                SyncRequestLog logDto = new SyncRequestLog();
                logDto.setMethod(method);
                logDto.setTargetUri(URI.create(template.url()));
                // 转换Feign headers为Spring HttpHeaders
                HttpHeaders headers = new HttpHeaders();
                template.headers().forEach((key, values) -> values.forEach(val -> headers.add(key, val)));
                logDto.setHeaders(headers);
                logDto.setRequestBody(template.body() != null ? objectMapper.readTree(template.body()) : null);

                // 异步发送到Kafka
                CompletableFuture.runAsync(() -> 
                    kafkaTemplate.send("d_new_sync_topic", objectMapper.writeValueAsString(logDto))
                ).exceptionally(ex -> {
                    log.error("Kafka send failed", ex);
                    return null;
                });
            } catch (JsonProcessingException e) {
                log.error("Request serialization failed", e);
            }
        }
    }

    // 同样的SyncRequestLog内部类
}

这个拦截器会自动作用于所有FeignClient，如果你只想针对<D_NEW>，可以在@FeignClient注解中指定configuration属性绑定这个拦截器。

方案3：Tomcat全局请求拦截（不推荐，除非需要全局拦截） #

如果你想在Tomcat层面拦截所有对外请求，也可以用Tomcat的Valve机制，但这个是全局生效的，需要额外判断请求是否发往<D_NEW>，而且要处理请求体的重复读取问题（因为Tomcat的请求流只能读一次）：

public class TomcatDataSyncValve extends ValveBase {

    private static final Logger log = LoggerFactory.getLogger(TomcatDataSyncValve.class);
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;
    // 替换为<D_NEW>的服务地址或标识
    private static final String TARGET_SERVICE_ID = "d-new-service";

    public TomcatDataSyncValve(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }

    @Override
    public void invoke(Request request, Response response) throws IOException, ServletException {
        HttpServletRequest servletRequest = (HttpServletRequest) request;
        // 包装请求，支持重复读取请求体
        ContentCachingRequestWrapper wrappedRequest = new ContentCachingRequestWrapper(servletRequest);

        // 判断是否是发往<D_NEW>的请求（根据Host或请求路径判断）
        String requestUrl = wrappedRequest.getRequestURL().toString();
        if (requestUrl.contains(TARGET_SERVICE_ID)) {
            String method = wrappedRequest.getMethod();
            if (Set.of("POST", "PUT", "DELETE").contains(method)) {
                try {
                    SyncRequestLog logDto = new SyncRequestLog();
                    logDto.setMethod(method);
                    logDto.setTargetUri(new URI(requestUrl));
                    // 读取请求头
                    HttpHeaders headers = new HttpHeaders();
                    Enumeration<String> headerNames = wrappedRequest.getHeaderNames();
                    while (headerNames.hasMoreElements()) {
                        String headerName = headerNames.nextElement();
                        headers.add(headerName, wrappedRequest.getHeader(headerName));
                    }
                    logDto.setHeaders(headers);
                    // 读取缓存的请求体
                    byte[] bodyBytes = wrappedRequest.getContentAsByteArray();
                    logDto.setRequestBody(bodyBytes.length > 0 ? objectMapper.readTree(bodyBytes) : null);

                    // 异步发送到Kafka
                    CompletableFuture.runAsync(() -> 
                        kafkaTemplate.send("d_new_sync_topic", objectMapper.writeValueAsString(logDto))
                    ).exceptionally(ex -> {
                        log.error("Kafka send failed", ex);
                        return null;
                    });
                } catch (Exception e) {
                    log.error("Request logging failed", e);
                }
            }
        }

        // 继续处理请求
        getNext().invoke(wrappedRequest, response);
    }
}

然后在Spring Boot中注册这个Valve：

@Configuration
public class TomcatConfig {

    @Bean
    public TomcatServletWebServerFactory tomcatFactory(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
        TomcatServletWebServerFactory factory = new TomcatServletWebServerFactory();
        factory.addEngineValves(new TomcatDataSyncValve(kafkaTemplate, objectMapper));
        return factory;
    }
}

注意事项： #

1. 异步发送Kafka：一定要用异步方式发送，避免阻塞主请求的处理流程，影响接口性能。
2. 请求体序列化：如果请求体是二进制格式（比如文件上传），需要特殊处理，不要强行转JSON。
3. 异常隔离：拦截器中的异常要自行处理，不要抛出到主业务流程中，避免因为Kafka故障导致正常请求失败。
4. 幂等性：后续重放请求到<D_OLD>时，要确保请求是幂等的，避免重复执行导致数据不一致。

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