likes
comments
collection
share

常见的限流手段和算法

作者站长头像
站长
· 阅读数 54

接口限流

我正在参加「掘金·启航计划」

中间件层面的限流处理

Tomcat：可以设置最大连接数，针对于单体的项目有效

Nginx：漏桶算法

Gateway：令牌桶算法

Nginx 限流

  1. 利用漏桶算法对请求进行限流
http {
  limit_req_zone $binary_remote_addr zone=servicelRateLimit:10m rate=10r/s
  server {
    listen 80;
    server_name localhost;
    location / {
      limit_req_zone servicelRateLimit burst=20 nodelay;
			proxy_pass http://targetserver;
    }
  }
}

语法：limit_req_zone key zone rate

  • key：定义限流对像，binary_remote_addr 就是一种key,基于客户端 ip 限流
  • Zone：定义共享存储区来存储访问信息，10m 可以存储 16wip 地址访问信息
  • Rate：最大访问速率，rate=10r/s 表示每秒最多请求 10 个请求
  • burst=20：相当于桶的大小
  • Nodelay：快速处理
  1. 控制并发的连接数
http {
  limit_conn_zone $binary_remote_addr zone=perip:10m;
  limit_conn_zone $server_name zone=perserver:10m;
  server {
    listen 80;
    server name localhost;
  	location / {
      limit conn perip 20;
      limit_conn perserver 100;
      proxy_pass http://targetserver;
  	}
  }
}
  • limit conn perip 20：对应的 key 是 $binary_remote_addr，表示限制单个lP同时最多能持有20个连接
  • limit_conn perserver 100：对应的 key 是 $server_name表示虚拟主机 (server) 同时能处理并发连接的总数

Gateway 限流

yml 配置文件中，微服务路由设置添加局部过滤器 RequestRateLimiter，基于的是令牌桶算法，默认使用 redis 存储令牌，需要配置 redis 的连接

- id:gateway-consumer
  uri:1b://GATEWAY-CONSUMER
  predicates:
  - Path=/order/**
  filters:
  - name:RequestRateLimiter
    args:
    	#使用SpEL从容器中获取对象
      key-resolver:'#@pathKeyResolver}'
      #令牌桶每秒填充平均速率
      redis-rate-limiter.replenishRate:1
      #令牌桶的上限
      redis-rate-limiter.burstCapacity:3
  • key-resolver：定义限流对像（ip、路径、参数)，需代码实现，使用 spel 表达式获取
  • redis-rate-limiter.replenishRate：令牌桶每秒填充平均速率
  • redis-rate-limiter.burstCapacity：令牌桶总容量。

Sentinel

Sentinel提供了丰富的功能特性，如流量控制、异常熔断、集群限流和速率控制等

常见的限流手段和算法

虽然Sentinel提供了丰富的功能特性，但我们当下需要重点关注的是流量控制部分。所谓流量控制，其原理是监控应用流量的 QPS 或并发线程数等指标，当达到指定的阈值时对流量进行控制，以避免被瞬时的流量高峰冲垮，从而保障应用的高可用性。

@GetMapping("/{activityId}/list/{itemId}")
@SentinelResource((value = "GetSeckillGood")
public BaseResponse<SeckillGoodResponse> getSeckillGood(@RequestHeader(value = "TokenInfo") Long userId,
														@PathVariable Long activityId,
														@PathVariable Long itemId,
														@RequestParam(required = false) Long version) {
	return seckillGoodService.getSeckillGood(userId, activityId, itemId, version);
}

请求接口添加 @SentinelResourse 接口

限流模式

单机限流

使用 Guava 中的单机限流工具即可

import com.google.common.util.concurrent.RateLimiter;

public class RateLimiterExample {
    public static void main(String[] args) {
        // 创建一个每秒允许2个请求的限流器
        RateLimiter rateLimiter = RateLimiter.create(2);

        // 模拟10个请求
        for (int i = 1; i <= 10; i++) {
            // 尝试获取令牌
            if (rateLimiter.tryAcquire()) {
                System.out.println("Request " + i + " is processed.");
            } else {
                System.out.println("Request " + i + " is rejected.");
            }
        }
    }
}

分布式限流

使用 Redis 记录用户的访问频率或者使用 Gateway 来进行统一的限流处理，这里展示使用 Redisson 自带的限流工具进行限流处理

/**
 * @author Ezreal
 * @Date 2023/6/22
 */
@Component
public class CurrentLimitManager {

    @Resource
    private RedissonClient redissonClient;

    public void doRateLimit(String key) {
        RRateLimiter rateLimiter = redissonClient.getRateLimiter(key);

        // 每秒钟最多访问两次
        rateLimiter.trySetRate(RateType.OVERALL, 2, 1, RateIntervalUnit.SECONDS);

        boolean acquire = rateLimiter.tryAcquire(1);
        if (!acquire) {
            throw new ToManyRequestException("to many request");
        }
    }
}

常见的限流算法

漏桶算法

设计一个漏桶，如果漏桶满了就可以拒绝服务，如果没有满，则可以通过固定的速率来处理漏桶中的请求

如果漏桶中没有水：

  • 如果进水速率小于等于最大出水速率，漏桶内不会有积水
  • 如果进水速率大于最大出水速率，漏桶内会产生积水

如果漏桶中存在水

  • 如果进水速率小于等于最大出水速率，那么漏桶内的水会被排干
  • 如果进水速率大于最大出水速率，那么漏桶中的水就会满，多于的水会溢出
/**
 * @author Ezreal
 * @Date 2023/6/22
 */
public class LeakyBucketWater {
    long lastModifyTime = 0L;

    long currentWater = 0L;

    long capacity;

    long rate = 2L;


    public LeakyBucketWater(long capacity) {
        this.capacity = capacity;
    }

    public Boolean doProcess() {
        long currentTimeMillis = System.currentTimeMillis();
        // 每分钟出水的个数，如何体现固定限流(currentTimeMillis - lastModifyTime) / 1000  取余的操作
        long outWater = (currentTimeMillis - lastModifyTime) / 1000 * rate;

        // 当前水的容量大小
        currentWater = Math.max(0, currentWater - outWater);

        if (currentWater < capacity) {
            lastModifyTime = currentTimeMillis;
            currentWater++;
            return true;
        } else {
            return false;
        }
    }
}

令牌桶算法

设计一个桶，以固定的速率向里面放入令牌，每次请求到来时，都会先领取令牌，再去执行相关的业务

与漏桶算法相比，令牌桶算法可以支持大量突发的请求，而漏桶算法处理的请求相对平滑

/**
 * @author Ezreal
 * @Date 2023/6/22
 */
public class TokenBucket {
    long lastModifyTime = 0L;

    long bucketCounts = 10L;

    long capacity = 50L;

    long currentBucket = 0;


    public Boolean doProcess() {
        long currentTimeMillis = System.currentTimeMillis();
        long generateBucket = (currentTimeMillis - lastModifyTime) / 1000 * bucketCounts;
        currentBucket = Math.min(capacity, generateBucket + currentBucket);
        lastModifyTime = currentTimeMillis;

        if (currentBucket > 0) {
            currentBucket--;
            return true;
        } else {
            return false;
        }
    }
}

基于 Redis 的滑动窗口限流算法

思路：

  1. 定义一个时间段的长度（即窗口长度 len）
  2. 统计[now - len, now] 之间请求的个数
  3. 若超过最大值，则直接返回错误信息即可；

使用 redis 中的 zset 来实现

  1. 使用用户的唯一标识（id、ip 等等）作为 key，当前时间 的作为 value，当前时间作为分数 score
  2. 当用户请求到来时，将当前的 key - value - score 加入到 zset 中（key 要设置过期时间）
  3. 计算 start 和 end 的值
  • end：now time
  • start：end - len
  1. 移除[0, start] 之间的标记
  2. 统计 [start, end] 之间 key 的数量，判断是否超过最大值即可
@Component
public class SlidingWindowLimitServiceImpl implements SlidingWindowLimitService {

    private final Long maxCount = 100L;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public boolean pass(String userKey, int period, int size) {
        int len = period * size;
        long now = System.currentTimeMillis();
        long start = now - len;
        // 将当前时间加入
        redisTemplate.opsForZSet().add(userKey, String.valueOf(now), now);
        redisTemplate.expire(userKey, len + period, TimeUnit.MILLISECONDS);
        // 移除 [0, start] 之间的记录
        redisTemplate.opsForZSet().reverseRangeByScore(userKey, 0, start);

        // 统计 (start, now] 的数量
        Long count = redisTemplate.opsForZSet().zCard(userKey);
        if (count == null) {
            return false;
        }

        return count <= maxCount;
    }
}
转载自:https://juejin.cn/post/7247451971945758780
评论
请登录