詳解Spring Cloud Gateway 限流操作

更新時間：2018年07月23日 13:47:02 作者：尹吉歡

這篇文章主要介紹了詳解Spring Cloud Gateway 限流操作,小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

開發(fā)高并發(fā)系統(tǒng)時有三把利器用來保護系統(tǒng)：緩存、降級和限流。

API網(wǎng)關作為所有請求的入口，請求量大，我們可以通過對并發(fā)訪問的請求進行限速來保護系統(tǒng)的可用性。

常用的限流算法比如有令牌桶算法，漏桶算法，計數(shù)器算法等。

在Zuul中我們可以自己去實現(xiàn)限流的功能（Zuul中如何限流在我的書《Spring Cloud微服務-全棧技術與案例解析》中有詳細講解），Spring Cloud Gateway的出現(xiàn)本身就是用來替代Zuul的。

要想替代那肯定得有強大的功能，除了性能上的優(yōu)勢之外，Spring Cloud Gateway還提供了很多新功能，比如今天我們要講的限流操作，使用起來非常簡單，今天我們就來學習在如何在Spring Cloud Gateway中進行限流操作。

目前限流提供了基于Redis的實現(xiàn),我們需要增加對應的依賴：

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

可以通過KeyResolver來指定限流的Key,比如我們需要根據(jù)用戶來做限流，IP來做限流等等。

IP限流

@Bean
public KeyResolver ipKeyResolver() {
 return exchange -> Mono.just(exchange.getRequest().getRemoteAddress().getHostName());
}

通過exchange對象可以獲取到請求信息，這邊用了HostName，如果你想根據(jù)用戶來做限流的話這邊可以獲取當前請求的用戶ID或者用戶名就可以了，比如：

用戶限流

使用這種方式限流，請求路徑中必須攜帶userId參數(shù)。

@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}"

filter名稱必須是RequestRateLimiter
redis-rate-limiter.replenishRate：允許用戶每秒處理多少個請求
redis-rate-limiter.burstCapacity：令牌桶的容量，允許在一秒鐘內(nèi)完成的最大請求數(shù)
key-resolver：使用SpEL按名稱引用bean

可以訪問接口進行測試，這時候Redis中會有對應的數(shù)據(jù)：

127.0.0.1:6379> keys *
1) "request_rate_limiter.{localhost}.timestamp"
2) "request_rate_limiter.{localhost}.tokens"

大括號中就是我們的限流Key,這邊是IP，本地的就是localhost

timestamp:存儲的是當前時間的秒數(shù)，也就是System.currentTimeMillis() / 1000或者Instant.now().getEpochSecond()
tokens:存儲的是當前這秒鐘的對應的可用的令牌數(shù)量

Spring Cloud Gateway目前提供的限流還是相對比較簡單的，在實際中我們的限流策略會有很多種情況，比如：

每個接口的限流數(shù)量不同，可以通過配置中心動態(tài)調(diào)整
超過的流量被拒絕后可以返回固定的格式給調(diào)用方
對某個服務進行整體限流（這個大家可以思考下用Spring Cloud Gateway如何實現(xiàn)，其實很簡單）
……

當然我們也可以通過重新RedisRateLimiter來實現(xiàn)自己的限流策略，這個我們后面再進行介紹。

限流源碼

// 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();

 // 令牌桶的容量，允許在一秒鐘內(nèi)完成的最大請求數(shù)
 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)
  // 執(zhí)行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腳本在：

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 }

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: