分布式系列之限流组件

/**
 * 指定过期时间自增计数器，默认每次+1，非滑动窗口
 *
 * @param key 计数器自增key
 * @param expireTime 过期时间
 * @param unit  时间单位
 */
public long incrCount(String key, int expireTime, TimeUnit unit);

/**
 * 指定过期时间自增计数器,单位时间内超过最大值rateThreshold返回true，否则返回false
 *
 * @param key 限流key
 * @param rateThreshold 限流阈值
 * @param expireTime 固定窗口时间
 * @param unit 时间单位
 */
public boolean rateLimit(final String key, final int rateThreshold, int expireTime, TimeUnit unit);

http {
    limit_conn_zone $binary_remote_addr zone=addr:10m;
    server {
    	location /download/ {
        	limit_conn addr 1;
        }	
    }
}

http {
    limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;
    server {
        location /search/ {
            limit_req zone=one burst=5;
        }
}

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId>
</dependency>
<!--基于 reactive stream 的redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>

spring:
  cloud:
    gateway:
      routes:
        - id: requestratelimiter_route
          uri: lb://gateway-demo
          order: 10000
          predicates:
          - Path = /admin/**
          filters:
          - name: RequestRateLimiter
            args:
              redis-rate-limiter.replenishRate: 1 # 流速每秒1个
              redis-rate-limiter.burstCapacity: 3 # 令牌桶容积
              key - resolver: "#{@remoteAddrKeyResolver}" # SPEL表达式去的对应的bean
          - StripPrefix = 1

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

public Mono<Response> isAllowed(String routeId, String id) {
    Config routeConfig = getConfig().getOrDefault(routeId, defaultConfig);
    int replenishRate = routeConfig.getReplenishRate();
    int burstCapacity = routeConfig.getBurstCapacity();
    try {
        List<String> keys = getKeys(id);
        /* returns unixtime in seconds.*/
        List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "", Instant.now().getEpochSecond() + "", "1");
        // 核心,执行redis LUA脚本
        Flux<List<Long>> flux = this.redisTemplate.execute(this.script, keys, scriptArgs);
        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)));
}

Bucket bucket = Bucket4j.builder().addLimit(limit).build();
if (bucket.tryConsume(1)) {
    log.info("ok");
} else {
    log.info("error");
}

Bandwidth limit = Bandwidth.simple(10, Duration.ofMinutes(1));

// 贪婪策略：桶大小为10，填充速度为每分钟5个令牌;
Refill filler = Refill.greedy(5, Duration.ofMinutes(1));
Bandwidth limit = Bandwidth.classic(10, filler);

// 创建一个Bulkhead,最大并发量为150
BulkheadConfig bulkheadConfig = BulkheadConfig.custom()
    .maxConcurrentCalls(150)
    .maxWaitTime(100)
    .build();
Bulkhead bulkhead = Bulkhead.of("backendName", bulkheadConfig);
 
// 创建RateLimiter，每秒允许一次请求
RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom()
    .timeoutDuration(Duration.ofMillis(100))
    .limitRefreshPeriod(Duration.ofSeconds(1))
    .limitForPeriod(1)
    .build();
RateLimiter rateLimiter = RateLimiter.of("backendName", rateLimiterConfig);
 
// 使用Bulkhead和RateLimiter装饰业务逻辑
Supplier<String> supplier = () -> backendService.doSomething();
Supplier<String> decoratedSupplier = Decorators.ofSupplier(supplier)
  .withBulkhead(bulkhead)
  .withRateLimiter(rateLimiter)
  .decorate();
 
// 调用业务逻辑
Try<String> try = Try.ofSupplier(decoratedSupplier);
assertThat(try.isSuccess()).isTrue();

@DoNotMock("Use FakeTimeLimiter")
@ElementTypesAreNonnullByDefault
@J2ktIncompatible
@GwtIncompatible
public interface TimeLimiter {
    <T> T newProxy(T target, Class<T> interfaceType, long timeoutDuration, TimeUnit timeoutUnit);

    default <T> T newProxy(T target, Class<T> interfaceType, Duration timeout) {
        return this.newProxy(target, interfaceType, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
    }

    @ParametricNullness
    @CanIgnoreReturnValue
    <T> T callWithTimeout(Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException, InterruptedException, ExecutionException;

    @ParametricNullness
    @CanIgnoreReturnValue
    default <T> T callWithTimeout(Callable<T> callable, Duration timeout) throws TimeoutException, InterruptedException, ExecutionException {
        return this.callWithTimeout(callable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
    }

    @ParametricNullness
    @CanIgnoreReturnValue
    <T> T callUninterruptiblyWithTimeout(Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException, ExecutionException;

    @ParametricNullness
    @CanIgnoreReturnValue
    default <T> T callUninterruptiblyWithTimeout(Callable<T> callable, Duration timeout) throws TimeoutException, ExecutionException {
        return this.callUninterruptiblyWithTimeout(callable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
    }

    void runWithTimeout(Runnable runnable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException, InterruptedException;

    default void runWithTimeout(Runnable runnable, Duration timeout) throws TimeoutException, InterruptedException {
        this.runWithTimeout(runnable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
    }

    void runUninterruptiblyWithTimeout(Runnable runnable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException;

    default void runUninterruptiblyWithTimeout(Runnable runnable, Duration timeout) throws TimeoutException {
        this.runUninterruptiblyWithTimeout(runnable, Internal.toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
    }
}

public void test() {
    TimeLimiter timeLimiter = SimpleTimeLimiter.create();
    Callable<String> callable = () -> {
        // Simulate long running task
        Thread.sleep(2000);
        return "Task Completed";
    };
    try {
        String result = timeLimiter.callWithTimeout(callable, 1, TimeUnit.SECONDS);
        System.out.println(result);
    } catch (UncheckedTimeoutException e) {
		// just logging or handle exception
    } catch (Exception e) {
	    // just logging or handle exception
    }
}