Gateway Filters: Глубокая архитектура и реактивная реализация
Реактивная природа фильтров в Spring Cloud Gateway
В Spring Cloud Gateway каждый фильтр реализует интерфейс GatewayFilter, который определяет единственный метод:
Ключевой аспект — возвращаемый тип Mono<Void>. Это реактивный тип, представляющий отложенное вычисление, которое может завершиться успешно, с ошибкой или никогда не завершиться. Фильтры не блокируют поток выполнения (thread) — вместо этого они описывают pipeline обработки, который выполняется асинхронно.
Пример неблокирующего фильтра:
Цепочка фильтров как реактивный pipeline
GatewayFilterChain представляет собой последовательность фильтров, организованную как связанный список. Когда вызывается chain.filter(exchange), выполнение передаётся следующему фильтру в цепочке. Последний фильтр в цепочке вызывает фактическую маршрутизацию запроса к целевому сервису через NettyRoutingFilter.
Внутренняя реализация DefaultGatewayFilterChain:
Важное наблюдение: цепочка фильтров не выполняется немедленно. Она описывается как последовательность операторов реактивного программирования, которые будут выполнены позже, когда на них подпишутся.
Порядок выполнения фильтров
Фильтры выполняются в строго определённом порядке:
GlobalFilter с наименьшим значением getOrder() (отрицательные значения имеют наивысший приоритет)
Остальные GlobalFilter в порядке возрастания значения getOrder()
Фильтры маршрута в порядке их объявления в конфигурации
Пример конфликта порядка:
#Java #middle #Spring_Cloud_Gateway #Filters
Реактивная природа фильтров в Spring Cloud Gateway
В Spring Cloud Gateway каждый фильтр реализует интерфейс GatewayFilter, который определяет единственный метод:
Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain);
Ключевой аспект — возвращаемый тип Mono<Void>. Это реактивный тип, представляющий отложенное вычисление, которое может завершиться успешно, с ошибкой или никогда не завершиться. Фильтры не блокируют поток выполнения (thread) — вместо этого они описывают pipeline обработки, который выполняется асинхронно.
Пример неблокирующего фильтра:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Этот код выполняется синхронно при построении цепочки
long startTime = System.currentTimeMillis();
// Возвращаем Mono, который описывает асинхронную операцию
return chain.filter(exchange)
.then(Mono.fromRunnable(() -> {
// Этот код выполняется асинхронно ПОСЛЕ завершения цепочки
long duration = System.currentTimeMillis() - startTime;
log.info("Request processed in {} ms", duration);
}));
}Цепочка фильтров как реактивный pipeline
GatewayFilterChain представляет собой последовательность фильтров, организованную как связанный список. Когда вызывается chain.filter(exchange), выполнение передаётся следующему фильтру в цепочке. Последний фильтр в цепочке вызывает фактическую маршрутизацию запроса к целевому сервису через NettyRoutingFilter.
Внутренняя реализация DefaultGatewayFilterChain:
public class DefaultGatewayFilterChain implements GatewayFilterChain {
private final List<GatewayFilter> filters;
private final int index;
private final Publisher<Void> currentPublisher;
@Override
public Mono<Void> filter(ServerWebExchange exchange) {
return Mono.defer(() -> {
if (this.index < filters.size()) {
GatewayFilter filter = filters.get(this.index);
DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(
this, this.index + 1);
return filter.filter(exchange, chain);
} else {
return Mono.empty(); // Конец цепочки
}
});
}
}Важное наблюдение: цепочка фильтров не выполняется немедленно. Она описывается как последовательность операторов реактивного программирования, которые будут выполнены позже, когда на них подпишутся.
Порядок выполнения фильтров
Фильтры выполняются в строго определённом порядке:
GlobalFilter с наименьшим значением getOrder() (отрицательные значения имеют наивысший приоритет)
Остальные GlobalFilter в порядке возрастания значения getOrder()
Фильтры маршрута в порядке их объявления в конфигурации
Пример конфликта порядка:
@Component
@Order(-1) // Выполняется ПЕРВЫМ
public class FirstGlobalFilter implements GlobalFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
log.info("First global filter");
return chain.filter(exchange);
}
}
@Component
@Order(0) // Выполняется ВТОРЫМ
public class SecondGlobalFilter implements GlobalFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
log.info("Second global filter");
return chain.filter(exchange);
}
}
#Java #middle #Spring_Cloud_Gateway #Filters
👍1
Детальный разбор встроенных фильтров
AddRequestHeader / AddResponseHeader
Эти фильтры добавляют заголовки к запросу или ответу.
Важно понимать их реактивную природу:
Особенность: Заголовки добавляются в момент модификации запроса, который происходит синхронно при построении цепочки. Однако сама отправка модифицированного запроса к целевому сервису происходит асинхронно.
RemoveRequestHeader
Удаляет заголовки из запроса.
Важная деталь — фильтр работает с копией заголовков:
RewritePath / SetPath
Эти фильтры модифицируют путь запроса. RewritePath использует регулярные выражения для замены, а SetPath устанавливает фиксированный путь.
Внутренняя механика RewritePath:
#Java #middle #Spring_Cloud_Gateway #Filters
AddRequestHeader / AddResponseHeader
Эти фильтры добавляют заголовки к запросу или ответу.
Важно понимать их реактивную природу:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
ServerHttpRequest request = exchange.getRequest();
// Модификация запроса (PRE-фильтр)
ServerHttpRequest mutatedRequest = request.mutate()
.header(headerName, headerValue)
.build();
// Создание нового обмена с модифицированным запросом
return chain.filter(exchange.mutate().request(mutatedRequest).build());
}Особенность: Заголовки добавляются в момент модификации запроса, который происходит синхронно при построении цепочки. Однако сама отправка модифицированного запроса к целевому сервису происходит асинхронно.
RemoveRequestHeader
Удаляет заголовки из запроса.
Важная деталь — фильтр работает с копией заголовков:
ServerHttpRequest mutated = exchange.getRequest().mutate()
.headers(headers -> {
headers.remove(headerName);
// или headers.removeIf(key -> key.startsWith("X-Secret-"));
})
.build();
RewritePath / SetPath
Эти фильтры модифицируют путь запроса. RewritePath использует регулярные выражения для замены, а SetPath устанавливает фиксированный путь.
Внутренняя механика RewritePath:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
String path = exchange.getRequest().getURI().getRawPath();
// Применение регулярного выражения
String newPath = path.replaceAll(regexp, replacement);
// Сохранение переменных из регулярного выражения
Map<String, String> variables = extractVariables(path, regexp);
exchange.getAttributes().put(URI_TEMPLATE_VARIABLES_ATTRIBUTE, variables);
// Построение нового URI
URI newUri = UriComponentsBuilder.fromUri(exchange.getRequest().getURI())
.replacePath(newPath)
.build(true)
.toUri();
ServerHttpRequest mutated = exchange.getRequest().mutate()
.uri(newUri)
.build();
return chain.filter(exchange.mutate().request(mutated).build());
}#Java #middle #Spring_Cloud_Gateway #Filters
👍1
RequestRateLimiter
Один из самых сложных фильтров, реализующий ограничение частоты запросов. Работает с Redis для распределённого лимитирования.
Архитектура:
Алгоритм token bucket в Redis:
#Java #middle #Spring_Cloud_Gateway #Filters
Один из самых сложных фильтров, реализующий ограничение частоты запросов. Работает с Redis для распределённого лимитирования.
Архитектура:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 1. Определение ключа для лимитирования
return keyResolver.resolve(exchange)
.switchIfEmpty(Mono.error(new IllegalStateException("Key not found")))
.flatMap(key -> {
// 2. Проверка лимита через Redis
return rateLimiter.isAllowed(key, config.getReplenishRate(),
config.getBurstCapacity(), config.getRequestedTokens())
.flatMap(response -> {
// 3. Добавление заголовков с информацией о лимите
exchange.getResponse().getHeaders()
.add("X-RateLimit-Remaining",
String.valueOf(response.getTokensLeft()));
exchange.getResponse().getHeaders()
.add("X-RateLimit-Replenish-Rate",
String.valueOf(config.getReplenishRate()));
if (response.isAllowed()) {
return chain.filter(exchange);
} else {
// 4. Отклонение запроса при превышении лимита
exchange.getResponse().setStatusCode(
HttpStatus.TOO_MANY_REQUESTS);
return exchange.getResponse().setComplete();
}
});
})
.onErrorResume(Exception.class, e -> {
// Обработка ошибок Redis
log.warn("Rate limiter error", e);
return chain.filter(exchange); // Разрешить запрос при ошибке
});
}Алгоритм token bucket в Redis:
-- Lua-скрипт, выполняемый атомарно в Redis
local tokens_key = KEYS[1] -- ключ для хранения токенов
local timestamp_key = KEYS[2] -- ключ для метки времени
local rate = tonumber(ARGV[1]) -- replenishRate
local capacity = tonumber(ARGV[2]) -- burstCapacity
local now = tonumber(ARGV[3]) -- текущее время
local requested = tonumber(ARGV[4]) -- requestedTokens
local fill_time = capacity / rate
local ttl = math.floor(fill_time * 2)
local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
last_tokens = capacity
end
local last_refreshed = tonumber(redis.call("get", timestamp_key))
if last_refreshed == nil then
last_refreshed = 0
end
local delta = math.max(0, now - last_refreshed)
local filled_tokens = math.min(capacity, last_tokens + (delta * rate))
local allowed = filled_tokens >= requested
local new_tokens = filled_tokens
if allowed then
new_tokens = filled_tokens - requested
end
redis.call("setex", tokens_key, ttl, new_tokens)
redis.call("setex", timestamp_key, ttl, now)
return { allowed, new_tokens }
#Java #middle #Spring_Cloud_Gateway #Filters
👍1
Retry / Circuit Breaker
Фильтры устойчивости, построенные на Resilience4J.
Retry фильтр:
Circuit Breaker фильтр:
ModifyResponseBody / ModifyRequestBody
Эти фильтры позволяют трансформировать тела запросов и ответов. Они требуют особого внимания из-за работы с потоком данных.
Архитектура ModifyRequestBody:
Критическое ограничение: Эти фильтры требуют чтения всего тела в память, что делает их непригодными для работы с большими файлами или стриминговыми данными.
#Java #middle #Spring_Cloud_Gateway #Filters
Фильтры устойчивости, построенные на Resilience4J.
Retry фильтр:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
return retryOperator.apply(chain.filter(exchange))
.onErrorResume(RetryExhaustedException.class, e -> {
// Все попытки исчерпаны
exchange.getResponse().setStatusCode(
HttpStatus.SERVICE_UNAVAILABLE);
return exchange.getResponse().setComplete();
});
}Circuit Breaker фильтр:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
return circuitBreaker.run(
chain.filter(exchange),
throwable -> {
// Fallback при открытом контуре или ошибке
ServerHttpResponse response = exchange.getResponse();
response.setStatusCode(HttpStatus.SERVICE_UNAVAILABLE);
response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
byte[] fallbackBody = "{\"status\":\"unavailable\"}".getBytes();
DataBuffer buffer = response.bufferFactory()
.wrap(fallbackBody);
return response.writeWith(Mono.just(buffer));
}
);
}ModifyResponseBody / ModifyRequestBody
Эти фильтры позволяют трансформировать тела запросов и ответов. Они требуют особого внимания из-за работы с потоком данных.
Архитектура ModifyRequestBody:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 1. Проверка Content-Type
MediaType contentType = exchange.getRequest().getHeaders().getContentType();
if (!config.getInClass().isInstance(contentType)) {
return chain.filter(exchange);
}
// 2. Чтение тела запроса в память (ограничение по size)
return DataBufferUtils.join(exchange.getRequest().getBody(), config.getMaxInMemorySize())
.switchIfEmpty(Mono.just(EMPTY_BUFFER))
.flatMap(dataBuffer -> {
// 3. Десериализация
return decode(dataBuffer, config.getInClass())
.flatMap(decodedBody -> {
// 4. Трансформация
return config.getRewriteFunction()
.apply(exchange, decodedBody)
.flatMap(encodedBody -> {
// 5. Сериализация обратно
byte[] bytes = encode(encodedBody, config.getOutClass());
// 6. Замена тела запроса
ServerHttpRequest mutated = exchange.getRequest().mutate()
.body(Flux.just(exchange.getResponse().bufferFactory()
.wrap(bytes)))
.build();
return chain.filter(
exchange.mutate().request(mutated).build()
);
});
})
.doFinally(signalType -> {
// 7. Освобождение буфера
DataBufferUtils.release(dataBuffer);
});
});
}Критическое ограничение: Эти фильтры требуют чтения всего тела в память, что делает их непригодными для работы с большими файлами или стриминговыми данными.
#Java #middle #Spring_Cloud_Gateway #Filters
👍1
Работа с телом запроса: проблемы и решения
Проблема повторного чтения body
Тело HTTP-запроса в реактивном стеке представлено как Flux<DataBuffer> — поток байтов, который может быть прочитан только один раз. При попытке повторного чтения возникает ошибка.
Неправильный подход:
Решения проблемы
1. Кэширование тела в памяти (для небольших запросов):
2. Использование ModifyRequestBody для трансформации:
3. Работа с потоком без буферизации (для больших данных):
#Java #middle #Spring_Cloud_Gateway #Filters
Проблема повторного чтения body
Тело HTTP-запроса в реактивном стеке представлено как Flux<DataBuffer> — поток байтов, который может быть прочитан только один раз. При попытке повторного чтения возникает ошибка.
Неправильный подход:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// ПЕРВОЕ чтение
return exchange.getRequest().getBody()
.collectList()
.flatMap(buffers -> {
// ВТОРОЕ чтение (ОШИБКА!)
return exchange.getRequest().getBody()
.collectList()
.flatMap(...);
});
}Решения проблемы
1. Кэширование тела в памяти (для небольших запросов):
@Component
public class CacheRequestBodyFilter implements GlobalFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Фильтр, который кэширует тело для последующего использования
return DataBufferUtils.join(exchange.getRequest().getBody())
.defaultIfEmpty(EMPTY_BUFFER)
.flatMap(dataBuffer -> {
// Сохраняем кэшированное тело в атрибутах
exchange.getAttributes().put(CACHED_REQUEST_BODY, dataBuffer);
// Создаём новый запрос с кэшированным телом
ServerHttpRequest mutated = exchange.getRequest().mutate()
.body(Flux.just(dataBuffer))
.build();
return chain.filter(exchange.mutate().request(mutated).build());
});
}
public static DataBuffer getCachedBody(ServerWebExchange exchange) {
return exchange.getAttribute(CACHED_REQUEST_BODY);
}
}
2. Использование ModifyRequestBody для трансформации:
filters:
- name: ModifyRequestBody
args:
inClass: String
outClass: String
rewriteFunction: "#{@myTransformer}"
3. Работа с потоком без буферизации (для больших данных):
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Преобразование потока на лету
Flux<DataBuffer> transformedBody = exchange.getRequest().getBody()
.map(dataBuffer -> {
// Трансформация каждого буфера
byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);
// Пример: преобразование в верхний регистр
String transformed = new String(bytes).toUpperCase();
return exchange.getResponse().bufferFactory()
.wrap(transformed.getBytes());
});
ServerHttpRequest mutated = exchange.getRequest().mutate()
.body(transformedBody)
.build();
return chain.filter(exchange.mutate().request(mutated).build());
}#Java #middle #Spring_Cloud_Gateway #Filters
👍1
Создание собственных фильтров
Реализация GatewayFilter через AbstractGatewayFilterFactory
Базовый фильтр с конфигурацией:
Использование в YAML:
Глобальные фильтры через GlobalFilter
Глобальные фильтры применяются ко всем маршрутам автоматически.
Пример: фильтр для добавления заголовка трассировки:
Фильтры с порядком выполнения
Порядок может быть задан через:
Аннотацию @Order
Реализацию интерфейса Ordered
Метод getOrder() в GatewayFilterFactory
Фильтр с динамическим порядком:
#Java #middle #Spring_Cloud_Gateway #Filters
Реализация GatewayFilter через AbstractGatewayFilterFactory
Базовый фильтр с конфигурацией:
@Component
public class LoggingGatewayFilterFactory extends
AbstractGatewayFilterFactory<LoggingGatewayFilterFactory.Config> {
public LoggingGatewayFilterFactory() {
super(Config.class);
}
@Override
public GatewayFilter apply(Config config) {
return (exchange, chain) -> {
// PRE-обработка
if (config.isLogRequest()) {
logRequest(exchange, config);
}
long startTime = System.nanoTime();
// POST-обработка через then()
return chain.filter(exchange)
.then(Mono.fromRunnable(() -> {
if (config.isLogResponse()) {
long duration = System.nanoTime() - startTime;
logResponse(exchange, config, duration);
}
}));
};
}
public static class Config {
private boolean logRequest = true;
private boolean logResponse = true;
private LogLevel level = LogLevel.INFO;
// Геттеры и сеттеры
}
@Override
public List<String> shortcutFieldOrder() {
return Arrays.asList("logRequest", "logResponse", "level");
}
}
Использование в YAML:
filters:
- Logging=true,true,DEBUG
Глобальные фильтры через GlobalFilter
Глобальные фильтры применяются ко всем маршрутам автоматически.
Пример: фильтр для добавления заголовка трассировки:
@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class TraceIdGlobalFilter implements GlobalFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Генерация или извлечение TraceId
String traceId = exchange.getRequest().getHeaders()
.getFirst("X-Trace-Id");
if (traceId == null || traceId.isEmpty()) {
traceId = UUID.randomUUID().toString();
}
// Добавление в заголовки запроса
ServerHttpRequest mutated = exchange.getRequest().mutate()
.header("X-Trace-Id", traceId)
.build();
// Добавление в атрибуты для использования в других фильтрах
exchange.getAttributes().put("traceId", traceId);
return chain.filter(exchange.mutate().request(mutated).build());
}
}
Фильтры с порядком выполнения
Порядок может быть задан через:
Аннотацию @Order
Реализацию интерфейса Ordered
Метод getOrder() в GatewayFilterFactory
Фильтр с динамическим порядком:
@Component
public class DynamicOrderFilter implements GlobalFilter, Ordered {
private final Environment environment;
private int order = 0;
public DynamicOrderFilter(Environment environment) {
this.environment = environment;
// Динамическое определение порядка из конфигурации
this.order = environment.getProperty("gateway.filter.order",
Integer.class, 0);
}
@Override
public int getOrder() {
return order;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Логика фильтра
return chain.filter(exchange);
}
}
#Java #middle #Spring_Cloud_Gateway #Filters
👍1
Сложный фильтр с асинхронными операциями
Пример: фильтр с внешним вызовом для валидации:
Фильтр с доступом к телу запроса
Безопасная работа с телом запроса:
#Java #middle #Spring_Cloud_Gateway #Filters
Пример: фильтр с внешним вызовом для валидации:
@Component
public class ExternalValidationFilter implements GlobalFilter {
private final WebClient webClient;
private final CircuitBreaker circuitBreaker;
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// 1. Извлечение данных для валидации
String apiKey = exchange.getRequest().getHeaders()
.getFirst("X-API-Key");
if (apiKey == null) {
return unauthorized(exchange);
}
// 2. Асинхронная валидация через внешний сервис
return circuitBreaker.run(
webClient.get()
.uri("/validate?apiKey={key}", apiKey)
.retrieve()
.bodyToMono(ValidationResponse.class)
.timeout(Duration.ofSeconds(5))
.onErrorResume(e -> Mono.just(new ValidationResponse(false))),
throwable -> Mono.just(new ValidationResponse(false))
)
.flatMap(validationResponse -> {
// 3. Обработка результата валидации
if (validationResponse.isValid()) {
// Добавление информации о пользователе в заголовки
ServerHttpRequest mutated = exchange.getRequest().mutate()
.header("X-User-Id", validationResponse.getUserId())
.header("X-User-Roles",
String.join(",", validationResponse.getRoles()))
.build();
return chain.filter(exchange.mutate().request(mutated).build());
} else {
return unauthorized(exchange);
}
});
}
private Mono<Void> unauthorized(ServerWebExchange exchange) {
exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
return exchange.getResponse().setComplete();
}
}
Фильтр с доступом к телу запроса
Безопасная работа с телом запроса:
@Component
public class RequestBodyLoggingFilter implements GlobalFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Только для определенных Content-Type
MediaType contentType = exchange.getRequest().getHeaders().getContentType();
if (contentType == null || !contentType.includes(MediaType.APPLICATION_JSON)) {
return chain.filter(exchange);
}
// Ограничение размера для логирования
final int maxBodySize = 1024 * 10; // 10KB
return DataBufferUtils.join(exchange.getRequest().getBody(), maxBodySize)
.defaultIfEmpty(EMPTY_BUFFER)
.flatMap(dataBuffer -> {
try {
// Чтение и логирование
byte[] bytes = new byte[Math.min(dataBuffer.readableByteCount(), maxBodySize)];
dataBuffer.read(bytes);
String body = new String(bytes, StandardCharsets.UTF_8);
log.debug("Request body (first {} bytes): {}", bytes.length, body);
// Восстановление тела
DataBuffer restoredBuffer = exchange.getResponse().bufferFactory()
.wrap(bytes);
// Продолжение цепочки с восстановленным телом
ServerHttpRequest mutated = exchange.getRequest().mutate()
.body(Flux.just(restoredBuffer)
.concatWith(exchange.getRequest().getBody()
.skipUntil(buffer -> false))) // Пропускаем уже прочитанное
.build();
return chain.filter(exchange.mutate().request(mutated).build());
} finally {
DataBufferUtils.release(dataBuffer);
}
});
}
}
#Java #middle #Spring_Cloud_Gateway #Filters
👍1
Оптимизация и best practices
Избегание блокирующих операций
Неправильно (блокирующий вызов):
Правильно (асинхронная обёртка):
Управление памятью и ресурсами
Освобождение DataBuffer:
Мониторинг и метрики фильтров
#Java #middle #Spring_Cloud_Gateway #Filters
Избегание блокирующих операций
Неправильно (блокирующий вызов):
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Блокирующий вызов в реактивном контексте
String result = blockingService.call(); // ОПАСНО!
return chain.filter(exchange);
}Правильно (асинхронная обёртка):
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
// Обёртка блокирующего вызова в отдельном пуле потоков
return Mono.fromCallable(() -> blockingService.call())
.subscribeOn(Schedulers.boundedElastic()) // Выделенный пул для блокирующих операций
.flatMap(result -> {
// Использование результата
return chain.filter(exchange);
});
}Управление памятью и ресурсами
Освобождение DataBuffer:
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
return DataBufferUtils.join(exchange.getRequest().getBody())
.flatMap(dataBuffer -> {
try {
// Работа с буфером
return processBuffer(dataBuffer)
.flatMap(result -> chain.filter(exchange));
} finally {
// Обязательное освобождение
DataBufferUtils.release(dataBuffer);
}
});
}Мониторинг и метрики фильтров
@Component
public class MetricsGlobalFilter implements GlobalFilter {
private final MeterRegistry meterRegistry;
private final Map<String, Timer> timerCache = new ConcurrentHashMap<>();
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
long startTime = System.nanoTime();
String routeId = exchange.getAttribute(ROUTE_ATTRIBUTE_ID);
return chain.filter(exchange)
.doOnSuccess(v -> recordSuccess(routeId, startTime))
.doOnError(e -> recordError(routeId, startTime, e));
}
private void recordSuccess(String routeId, long startTime) {
long duration = System.nanoTime() - startTime;
Timer timer = timerCache.computeIfAbsent(routeId, id ->
Timer.builder("gateway.request.duration")
.tag("route", id)
.tag("status", "success")
.register(meterRegistry));
timer.record(duration, TimeUnit.NANOSECONDS);
}
}
#Java #middle #Spring_Cloud_Gateway #Filters
👍1