Spring 具有 Spring 缓存和咖啡因的云网关

Question

我有一个 spring 云网关，可以将 API 休息请求转发给一些微服务。

我想缓存特定请求的响应。为此，我写了这个 Filter

@Component
@Slf4j
public class CacheResponseGatewayFilterFactory extends AbstractGatewayFilterFactory<CacheResponseGatewayFilterFactory.Config> {
   private final CacheManager cacheManager;

   public CacheResponseGatewayFilterFactory(CacheManager cacheManager) {
     super(CacheResponseGatewayFilterFactory.Config.class);
     this.cacheManager = cacheManager;
   }

   @Override
   public GatewayFilter apply(CacheResponseGatewayFilterFactory.Config config) {
     final var cache = cacheManager.getCache("MyCache");
     return (exchange, chain) -> {
        final var path = exchange.getRequest().getPath();
        if (nonNull(cache.get(path))) {
            log.info("Return cached response for request: {}", path);
            final var response = cache.get(path, ServerHttpResponse.class);
            final var mutatedExchange = exchange.mutate().response(response).build();
            return mutatedExchange.getResponse().setComplete();
        }

        return chain.filter(exchange).doOnSuccess(aVoid -> {
            cache.put(path, exchange.getResponse());
        });
    };
}

当我调用我的rest端点时，第一次收到正确的json，第二次收到一个空体。

我做错了什么？

编辑这是 exchange.getRequest() 在执行 cache.put()

之前的屏幕截图

Answer 1

我解决了创建 GlobalFilter 和 ServerHttpResponseDecorator 的问题。无论如何，此代码都会缓存所有响应（可以轻松改进以仅缓存特定响应）。

这是代码。但是我认为它可以改进。万一让我知道呢。

@Slf4j
@Component
public class CacheFilter implements GlobalFilter, Ordered {
  private final CacheManager cacheManager;

  public CacheFilter(CacheManager cacheManager) {
    this.cacheManager = cacheManager;
  }

  @Override
  public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    final var cache = cacheManager.getCache("MyCache");

    final var cachedRequest = getCachedRequest(exchange.getRequest());
    if (nonNull(cache.get(cachedRequest))) {
        log.info("Return cached response for request: {}", cachedRequest);
        final var cachedResponse = cache.get(cachedRequest, CachedResponse.class);

        final var serverHttpResponse = exchange.getResponse();
        serverHttpResponse.setStatusCode(cachedResponse.httpStatus);
        serverHttpResponse.getHeaders().addAll(cachedResponse.headers);
        final var buffer = exchange.getResponse().bufferFactory().wrap(cachedResponse.body);
        return exchange.getResponse().writeWith(Flux.just(buffer));
    }

    final var mutatedHttpResponse = getServerHttpResponse(exchange, cache, cachedRequest);
    return chain.filter(exchange.mutate().response(mutatedHttpResponse).build());
  }

  private ServerHttpResponse getServerHttpResponse(ServerWebExchange exchange, Cache cache, CachedRequest cachedRequest) {
    final var originalResponse = exchange.getResponse();
    final var dataBufferFactory = originalResponse.bufferFactory();

    return new ServerHttpResponseDecorator(originalResponse) {

        @NonNull
        @Override
        public Mono<Void> writeWith(@NonNull Publisher<? extends DataBuffer> body) {
            if (body instanceof Flux) {
                final var flux = (Flux<? extends DataBuffer>) body;
                return super.writeWith(flux.buffer().map(dataBuffers -> {
                    final var outputStream = new ByteArrayOutputStream();
                    dataBuffers.forEach(dataBuffer -> {
                        final var responseContent = new byte[dataBuffer.readableByteCount()];
                        dataBuffer.read(responseContent);
                        try {
                            outputStream.write(responseContent);
                        } catch (IOException e) {
                            throw new RuntimeException("Error while reading response stream", e);
                        }
                    });
                    if (Objects.requireNonNull(getStatusCode()).is2xxSuccessful()) {
                        final var cachedResponse = new CachedResponse(getStatusCode(), getHeaders(), outputStream.toByteArray());
                        log.debug("Request {} Cached response {}", cacheKey.getPath(), new String(cachedResponse.getBody(), UTF_8));
                        cache.put(cacheKey, cachedResponse);
                    }
                    return dataBufferFactory.wrap(outputStream.toByteArray());
                }));
            }
            return super.writeWith(body);
        }
    };
  }

  @Override
  public int getOrder() {
    return -2;
  }

  private CachedRequest getCachedRequest(ServerHttpRequest request) {
    return CachedRequest.builder()
            .method(request.getMethod())
            .path(request.getPath())
            .queryParams(request.getQueryParams())
            .build();
  }

  @Value
  @Builder
  private static class CachedRequest {
    RequestPath path;
    HttpMethod method;
    MultiValueMap<String, String> queryParams;

  }

  @Value
  private static class CachedResponse {
    HttpStatus httpStatus;
    HttpHeaders headers;
    byte[] body;
  }
}

Spring 具有 Spring 缓存和咖啡因的云网关

Spring cloud gateway with Spring cache and caffeine

spring

spring-boot

spring-cache

spring-cloud-gateway

caffeine-cache