背景

排查 & 解决

RequestRateLimiterGatewayFilter

1. 通过指定的KeyResolver从请求里解析出限流key

2. 通过指定的RateLimiter来判定该请求应该被放行or被限流

3. 如放行则filterChain继续往下走，如限流则直接返回响应码429（可配置）

KeyResolver

public class UrlWithUserParamKeyResolver implements KeyResolver {

    @Override
    public Mono<String> resolve(ServerWebExchange exchange) {
        ServerHttpRequest request = exchange.getRequest();
        String authorization = request.getHeaders().getFirst("Authorization");
        if (StringUtils.isBlank(authorization)) {
            return Mono.justOrEmpty(authorization);
        }
        String path = request.getURI().getPath();
        String sign = request.getQueryParams().getFirst("sign");
        if (StringUtils.isBlank(sign)) {
            return Mono.justOrEmpty(authorization);
        }
        return Mono.just(path + "|" + sign);
    }
}

RateLimiter

1. replenishRate 填充速率，也就是每秒可以往桶里补充的令牌数

2. burstCapacity 桶容量，到达这个数量令牌就补充不进去了

3. requestedTokens 请求消耗数量，每次请求消耗多少令牌

接入RedisRateLimiter

引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>

自定义逻辑

1. 支持通过开关来控制限流逻辑

2. 支持通过开关来控制限流后是真正限流还是只打印观测日志

3. 调用redis执行限流脚本支持超时配置，超时后放行，不影响业务

4. 支持被限流后的异常文案可配置化

@Slf4j
@ConfigurationProperties("com.xxx.gateway.filter.request-rate-limiter")
@Component
public class RequestRateLimiter2GatewayFilterFactory extends
        AbstractGatewayFilterFactory<RequestRateLimiter2GatewayFilterFactory.Config> {

	/**
	 * Key-Resolver key.
	 */
	public static final String KEY_RESOLVER_KEY = "keyResolver";

	private static final String EMPTY_KEY = "____EMPTY_KEY__";

	private final RateLimiter defaultRateLimiter;

	private final KeyResolver defaultKeyResolver;

	/**
	 * Switch to deny requests if the Key Resolver returns an empty key, defaults to true.
	 */
	private boolean denyEmptyKey = true;

	/** HttpStatus to return when denyEmptyKey is true, defaults to FORBIDDEN. */
	private String emptyKeyStatusCode = HttpStatus.FORBIDDEN.name();

	public RequestRateLimiter2GatewayFilterFactory(RateLimiter defaultRateLimiter,
                                                   KeyResolver defaultKeyResolver) {
		super(Config.class);
		this.defaultRateLimiter = defaultRateLimiter;
		this.defaultKeyResolver = defaultKeyResolver;
	}

	@SuppressWarnings("unchecked")
	@Override
	public GatewayFilter apply(Config config) {
		KeyResolver resolver = getOrDefault(config.keyResolver, defaultKeyResolver);
		RateLimiter<Object> limiter = getOrDefault(config.rateLimiter,
				defaultRateLimiter);
		boolean denyEmpty = getOrDefault(config.denyEmptyKey, this.denyEmptyKey);
		HttpStatusHolder emptyKeyStatus = HttpStatusHolder
				.parse(getOrDefault(config.emptyKeyStatus, this.emptyKeyStatusCode));

		return (exchange, chain) -> {
			if (!config.isOpen()) {
				return chain.filter(exchange);
			}
			return resolver.resolve(exchange).defaultIfEmpty(EMPTY_KEY)
					.flatMap(key -> {
						if (EMPTY_KEY.equals(key)) {
							if (denyEmpty) {
								setResponseStatus(exchange, emptyKeyStatus);
								return exchange.getResponse().setComplete();
							}
							return chain.filter(exchange);
						}
						String routeId = config.getRouteId();
						if (routeId == null) {
							Route route = exchange
									.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);
							routeId = route.getId();
						}
						long start = System.currentTimeMillis();
						return limiter.isAllowed(routeId, key).timeout(Duration.ofMillis(config.getTimeout())).onErrorResume(throwable -> {
							long end = System.currentTimeMillis();
							log.warn("rate limit occurs error, cost:{} ms", (end - start), throwable);
							return Mono.just(new RateLimiter.Response(true, Collections.emptyMap()));
						}).flatMap(response -> {
							long end = System.currentTimeMillis();
							log.info("rate limit lua script cost:{} ms", (end - start));
							for (Map.Entry<String, String> header : response.getHeaders()
									.entrySet()) {
								exchange.getResponse().getHeaders().add(header.getKey(),
										header.getValue());
							}

							if (response.isAllowed()) {
								return chain.filter(exchange);
							}

							if (!config.isBlock()) {
								exchange.getResponse().getHeaders().add("x-exceed-unblock", "1");
								log.info("did not block, but record, key:{}, userId:{}", key, exchange.getRequest().getHeaders().getFirst("x-user-id"));
								return chain.filter(exchange);
							}

//						setResponseStatus(exchange, config.getStatusCode());
							String text = "{\\"error\\":{\\"code\\":\\"%s\\",\\"message\\":\\"%s\\",\\"detail\\":\\"\\",\\"exception\\":\\"\\"}}";
							return ExceptionUtils.exceedLimit(exchange, config.getStatusCode(), String.format(text, config.getStatusCode().value(), config.getMsg()));
						});
					});
		};
	}

	private <T> T getOrDefault(T configValue, T defaultValue) {
		return (configValue != null) ? configValue : defaultValue;
	}

	public static class Config implements HasRouteId {

		private KeyResolver keyResolver;

		private RateLimiter rateLimiter;

		private HttpStatus statusCode = HttpStatus.TOO_MANY_REQUESTS;

		private Boolean denyEmptyKey;

		private String emptyKeyStatus;

		private String routeId;

		private String msg = "请求过于频繁，请稍后再试";

		private boolean block;

		private boolean open;

		private long timeout = 100;
	}
}

配置

com.xxx.gateway.filter.request-rate-limiter.deny-empty-key=false
com.xxx.gateway.filter.request-rate-limiter.empty-key-status-code=429

spring.cloud.gateway.routes[0].id=test_host_route
spring.cloud.gateway.routes[0].uri=glb://test-service
spring.cloud.gateway.routes[0].predicates[0].name=Host
spring.cloud.gateway.routes[0].predicates[0].args.patterns[0]=test.com
spring.cloud.gateway.routes[0].filters[0].name=RequestRateLimiter2
# 控制是否打开限流逻辑
spring.cloud.gateway.routes[0].filters[0].args.open=true
# 控制达到限流阈值后是阻塞请求还是打印观测日志
spring.cloud.gateway.routes[0].filters[0].args.block=true
# 控制限流逻辑执行超时时间
spring.cloud.gateway.routes[0].filters[0].args.timeout=150
spring.cloud.gateway.routes[0].filters[0].args.redis-rate-limiter.replenishRate=5
spring.cloud.gateway.routes[0].filters[0].args.redis-rate-limiter.burstCapacity=5
spring.cloud.gateway.routes[0].filters[0].args.redis-rate-limiter.requestedTokens=1
spring.cloud.gateway.routes[0].filters[1].name=Retry
spring.cloud.gateway.routes[0].filters[1].args.retries=1
spring.cloud.gateway.routes[0].filters[1].args.methods=GET,POST

• https://stackoverflow.com/questions/51836375/redis-client-lettuce-command-timeout-versus-socket-timeout

后续优化思路

• 比如爬虫来遍历整站数据

• 比如DDos攻击

1. 我们可以做多维度的限流器，比如上面说的是 用户 + 请求 + 参数 维度。那我们还可以针对用户维度，限制一段时间的请求量

2. 可以增加恶意流量识别的逻辑，并针对恶意用户，做用户维度或者IP维度的封禁。再结合产品化的解禁策略，比如图形验证码等等

限流脚本代码分析

local tokens_key = KEYS[1]
local timestamp_key = KEYS[2]

local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

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
local allowed_num = 0
if allowed then
  new_tokens = filled_tokens - requested
  allowed_num = 1
end

if ttl > 0 then
  redis.call("setex", tokens_key, ttl, new_tokens)
  redis.call("setex", timestamp_key, ttl, now)
end

return { allowed_num, new_tokens }

存在的问题

redis.replicate_commands

redis.replicate_commands()

一点优化

参考

1. Spring Cloud RedisRateLimit限频存在的几个问题

2. Redis · 引擎特性 · Lua脚本新姿势