開發高併發系統時有三把利器用來保護系統:緩存、降級和限流。react
API網關做爲全部請求的入口,請求量大,咱們能夠經過對併發訪問的請求進行限速來保護系統的可用性。redis
經常使用的限流算法好比有令牌桶算法,漏桶算法,計數器算法等。算法
在Zuul中咱們能夠本身去實現限流的功能(Zuul中如何限流在個人書《Spring Cloud微服務-全棧技術與案例解析》中有詳細講解),Spring Cloud Gateway的出現自己就是用來替代Zuul的。spring
要想替代那確定得有強大的功能,除了性能上的優點以外,Spring Cloud Gateway還提供了不少新功能,好比今天咱們要講的限流操做,使用起來很是簡單,今天咱們就來學習在如何在Spring Cloud Gateway中進行限流操做。api
目前限流提供了基於Redis的實現,咱們須要增長對應的依賴:緩存
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis-reactive</artifactId> </dependency>
能夠經過KeyResolver來指定限流的Key,好比咱們須要根據用戶來作限流,IP來作限流等等。微信
IP限流併發
@Bean public KeyResolver ipKeyResolver() { return exchange -> Mono.just(exchange.getRequest().getRemoteAddress().getHostName()); }
經過exchange對象能夠獲取到請求信息,這邊用了HostName,若是你想根據用戶來作限流的話這邊能夠獲取當前請求的用戶ID或者用戶名就能夠了,好比:ide
用戶限流
使用這種方式限流,請求路徑中必須攜帶userId參數。spring-boot
@Bean KeyResolver userKeyResolver() { return exchange -> Mono.just(exchange.getRequest().getQueryParams().getFirst("userId")); }
接口限流
獲取請求地址的uri做爲限流key。
@Bean KeyResolver apiKeyResolver() { return exchange -> Mono.just(exchange.getRequest().getPath().value()); }
而後配置限流的過濾器信息:
server: port: 8084 spring: redis: host: 127.0.0.1 port: 6379 cloud: gateway: routes: - id: fsh-house uri: lb://fsh-house predicates: - Path=/house/** filters: - name: RequestRateLimiter args: redis-rate-limiter.replenishRate: 10 redis-rate-limiter.burstCapacity: 20 key-resolver: "#{@ipKeyResolver}"
127.0.0.1:6379> keys * 1) "request_rate_limiter.{localhost}.timestamp" 2) "request_rate_limiter.{localhost}.tokens"
大括號中就是咱們的限流Key,這邊是IP,本地的就是localhost
Spring Cloud Gateway目前提供的限流仍是相對比較簡單的,在實際中咱們的限流策略會有不少種狀況,好比:
// routeId也就是咱們的fsh-house,id就是限流的key,也就是localhost。 public Mono<Response> isAllowed(String routeId, String id) { // 會判斷RedisRateLimiter是否初始化了 if (!this.initialized.get()) { throw new IllegalStateException("RedisRateLimiter is not initialized"); } // 獲取routeId對應的限流配置 Config routeConfig = getConfig().getOrDefault(routeId, defaultConfig); if (routeConfig == null) { throw new IllegalArgumentException("No Configuration found for route " + routeId); } // 容許用戶每秒作多少次請求 int replenishRate = routeConfig.getReplenishRate(); // 令牌桶的容量,容許在一秒鐘內完成的最大請求數 int burstCapacity = routeConfig.getBurstCapacity(); try { // 限流key的名稱(request_rate_limiter.{localhost}.timestamp,request_rate_limiter.{localhost}.tokens) List<String> keys = getKeys(id); // The arguments to the LUA script. time() returns unixtime in seconds. List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "", Instant.now().getEpochSecond() + "", "1"); // allowed, tokens_left = redis.eval(SCRIPT, keys, args) // 執行LUA腳本 Flux<List<Long>> flux = this.redisTemplate.execute(this.script, keys, scriptArgs); // .log("redisratelimiter", Level.FINER); return flux.onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L))) .reduce(new ArrayList<Long>(), (longs, l) -> { longs.addAll(l); return longs; }) .map(results -> { boolean allowed = results.get(0) == 1L; Long tokensLeft = results.get(1); Response response = new Response(allowed, getHeaders(routeConfig, tokensLeft)); if (log.isDebugEnabled()) { log.debug("response: " + response); } return response; }); } catch (Exception e) { log.error("Error determining if user allowed from redis", e); } return Mono.just(new Response(true, getHeaders(routeConfig, -1L))); }
LUA腳本在:
WX20180715-150447@2x.png
local tokens_key = KEYS[1] local timestamp_key = KEYS[2] --redis.log(redis.LOG_WARNING, "tokens_key " .. tokens_key) 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) --redis.log(redis.LOG_WARNING, "rate " .. ARGV[1]) --redis.log(redis.LOG_WARNING, "capacity " .. ARGV[2]) --redis.log(redis.LOG_WARNING, "now " .. ARGV[3]) --redis.log(redis.LOG_WARNING, "requested " .. ARGV[4]) --redis.log(redis.LOG_WARNING, "filltime " .. fill_time) --redis.log(redis.LOG_WARNING, "ttl " .. ttl) local last_tokens = tonumber(redis.call("get", tokens_key)) if last_tokens == nil then last_tokens = capacity end --redis.log(redis.LOG_WARNING, "last_tokens " .. last_tokens) local last_refreshed = tonumber(redis.call("get", timestamp_key)) if last_refreshed == nil then last_refreshed = 0 end --redis.log(redis.LOG_WARNING, "last_refreshed " .. last_refreshed) 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 --redis.log(redis.LOG_WARNING, "delta " .. delta) --redis.log(redis.LOG_WARNING, "filled_tokens " .. filled_tokens) --redis.log(redis.LOG_WARNING, "allowed_num " .. allowed_num) --redis.log(redis.LOG_WARNING, "new_tokens " .. new_tokens) redis.call("setex", tokens_key, ttl, new_tokens) redis.call("setex", timestamp_key, ttl, now) return { allowed_num, new_tokens }
猿天地
點擊圖片查看更多推薦內容
↓↓↓
Spring Cloud Gateway 網關嚐鮮
Spring Cloud Gateway Eureka路由轉發
大牛坐鎮|高端JAVA純技術羣你要加入嗎?
更多技術分享盡在微信羣,加羣請關注公衆號,點擊加羣按鈕。
尹吉歡我不差錢啊稀罕做者1 人喜歡