4개의 분산 전류 제한 알고리즘 및 코드 구현
- Java后端技术全栈앞으로
- 2023-08-15 15:54:211139검색
질문으로 전류 한도에 접근
왜 전류를 제한해야 하나요?
위에서 말했듯이 트래픽이 많아지면 좋은 일이지만, 과부하가 걸리고 시스템이 정지되면 모두가 고통을 겪게 됩니다.
따라서 다양한 주요 프로모션 활동을 시작하기 전에 시스템에 대한 스트레스 테스트를 수행하고 전체 시스템의 최대 QPS를 평가하며 이를 초과하는 경우 일부 전류 제한 설정을 지정해야 합니다. 특정 임계값 이상인 경우 시스템 중단을 방지하기 위해 처리를 거부하거나 처리를 지연합니다.
전류 제한과 퓨즈의 차이점은 무엇인가요?
트래픽이 들어오기 전에 트래픽 제한이 발생하여 초과 트래픽이 제한됩니다.
회로 차단기는 트래픽이 들어온 후에 발생하는 오류를 처리하는 메커니즘입니다. 시스템에 오류가 발생하거나 비정상적인 경우 퓨즈는 오류가 더 확대되어 서비스 사태가 발생하는 것을 방지하기 위해 자동으로 요청을 차단합니다.
전류 제한과 피크 클리핑의 차이점은 무엇인가요?
피크 셰이빙은 트래픽을 원활하게 처리하는 프로세스로, 요청 처리 속도를 천천히 높여 시스템의 즉각적인 과부하를 방지합니다.
피크 커팅은 아마도 흐름을 저장하고 천천히 흐르는 저장소일 것입니다. 흐름 제한은 아마도 과잉 흐름을 거부하는 게이트일 것입니다.
전류 제한의 일반적인 프로세스
그럼 특정 전류 제한을 구현하는 방법은 무엇일까요? 다음 단계로 요약할 수 있습니다.
통계적 요청 트래픽: 카운터, 슬라이딩 창 등을 통해 계산할 수 있는 요청 수 또는 비율을 기록합니다. 제한 초과 여부 확인: 설정된 제한 조건에 따라 현재 요청 트래픽이 제한을 초과하는지 확인합니다. 현재 제한 전략 실행: 요청 트래픽이 제한을 초과하는 경우 요청 거부, 처리 지연, 오류 정보 반환 등의 현재 제한 전략을 구현합니다. 통계 정보 업데이트: 카운터 값 증가, 슬라이딩 윈도우의 데이터 업데이트 등 요청 처리 결과를 기반으로 통계 정보를 업데이트합니다. 위 단계 반복: 요청 트래픽을 지속적으로 계산하고, 한도 초과 여부를 확인하고, 현재 제한 정책을 구현하고, 통계 정보를 업데이트합니다.
특정 전류 제한 알고리즘 구현에 유의해야 합니다. 토큰 버킷 알고리즘, 누출 버킷 알고리즘 등을 사용하는 등 시나리오와 필요에 따라 조정하고 최적화합니다.
단일 머신 전류 제한 및 분산 전류 제한
전류 제한의 일반적인 프로세스에서는 요청 볼륨을 계산하고 통계를 업데이트해야 하므로 요청 볼륨의 통계 및 업데이트는 다음과 같아야 합니다. 저장소에 보관됩니다.
그냥 독립형 환경이라면 로컬에서 직접 처리하고 보관하기 쉽습니다.
그러나 일반적으로 우리 서비스는 클러스터에 배포됩니다. 여러 머신 간에 전반적인 전류 제한을 달성하는 방법은 무엇입니까?
이때 우리는 Tair나 Redis와 같은 분산형 K-V 스토리지에 통계 정보를 넣을 수 있습니다.
네 가지 전류 제한 알고리즘 및 분산 구현
다음으로 몇 가지 일반적인 전류 제한 알고리즘을 구현하기 시작합니다. 여기서는 Redis를 분산 저장소로 사용합니다. 말할 것도 없이 Redis는 Redis 클라이언트로서 가장 널리 사용되는 분산 캐시 DB입니다. Redission은 단순히 분산 잠금을 수행하는 데 사용되는데, 이는 다소 "어리석은" 일이지만 실제로 Redis 클라이언트로 사용하기도 매우 쉽습니다.
시작하기 전에 Redis 설치 및 프로젝트 생성에 대한 자세한 내용은 다루지 않겠습니다.
종속성 추가
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.2</version>
</dependency>RedissonClient를 얻으려면 싱글톤 모드를 사용하세요. 여기서는 Bean으로 등록되지 않습니다. 단일 테스트 실행이 너무 느립니다.
public class RedissonConfig {
private static final String REDIS_ADDRESS = "redis://127.0.0.1:6379";
private static volatile RedissonClient redissonClient;
public static RedissonClient getInstance(){
if (redissonClient==null){
synchronized (RedissonConfig.class){
if (redissonClient==null){
Config config = new Config();
config.useSingleServer().setAddress(REDIS_ADDRESS);
redissonClient = Redisson.create(config);
return redissonClient;
}
}
}
return redissonClient;
}
}창 전류 제한 알고리즘 수정
알고리즘 원리
고정 창 알고리즘은 많은 참고 자료에서 카운터 알고리즘이라고도 합니다. 물론 저는 개인적으로 카운터 알고리즘이 고정 창 알고리즘의 특별한 경우라는 것을 알고 있습니다. 물론 우리는 그것에 대해 크게 걱정하지 않습니다. .
고정 창 알고리즘은 비교적 간단한 전류 제한 알고리즘으로 시간을 고정 시간 창으로 나누고 각 창에서 허용되는 요청 수에 대한 제한을 설정합니다. 요청 수가 특정 기간 내 상한을 초과하면 전류 제한이 트리거됩니다.
알고리즘 구현
Redisson 기반의 고정 창 구현은 매우 간단합니다. 각 창 기간 내에서 incrementAndGet 작업을 통해 요청 수를 계산할 수 있습니다. 기간이 끝나면 Redis의 키 만료 기능을 활용하여 자동으로 개수를 재설정할 수 있습니다.
코드 구현을 살펴보겠습니다.
public class FixedWindowRateLimiter {
public static final String KEY = "fixedWindowRateLimiter:";
/**
* 请求限制数量
*/
private Long limit;
/**
* 窗口大小（单位：S）
*/
private Long windowSize;
public FixedWindowRateLimiter(Long limit, Long windowSize) {
this.limit = limit;
this.windowSize = windowSize;
}
/**
* 固定窗口限流
*/
public boolean triggerLimit(String path) {
RedissonClient redissonClient = RedissonConfig.getInstance();
//加分布式锁，防止并发情况下窗口初始化时间不一致问题
RLock rLock = redissonClient.getLock(KEY + "LOCK:" + path);
try {
rLock.lock(100, TimeUnit.MILLISECONDS);
String redisKey = KEY + path;
RAtomicLong counter = redissonClient.getAtomicLong(redisKey);
//计数
long count = counter.incrementAndGet();
//如果为1的话，就说明窗口刚初始化
if (count == 1) {
//直接设置过期时间，作为窗口
counter.expire(windowSize, TimeUnit.SECONDS);
}
//触发限流
if (count > limit) {
//触发限流的不记在请求数量中
counter.decrementAndGet();
return true;
}
return false;
} finally {
rLock.unlock();
}
}
}여기에서는 동시 상황에서 창 초기화 문제를 해결하기 위해 추가 분산 잠금이 사용됩니다.
다시 테스트해보자
class FixedWindowRateLimiterTest {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(20, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
@Test
@DisplayName("1min限制10次请求固定窗口测试")
void triggerLimit() throws InterruptedException {
FixedWindowRateLimiter fixedWindowRateLimiter = new FixedWindowRateLimiter(10L,60L);
//模拟不同窗口内的调用
for (int i = 0; i < 3; i++) {
CountDownLatch countDownLatch = new CountDownLatch(20);
//20个线程并发调用
for (int j = 0; j < 20; j++) {
threadPoolExecutor.execute(() -> {
boolean isLimit = fixedWindowRateLimiter.triggerLimit("/test");
System.out.println(isLimit);
countDownLatch.countDown();
});
}
countDownLatch.await();
//休眠1min
TimeUnit.MINUTES.sleep(1);
}
}
}물론 인터페이스를 작성하고 Jmeter와 같은 스트레스 테스트 도구를 사용하여 테스트할 수도 있습니다.
고정 창 알고리즘의 장점은 구현이 간단하고 공간을 적게 차지한다는 점이지만 창 전환이 즉시 완료되므로 요청 처리가 원활하지 않고 트래픽의 급격한 변동이 발생할 수 있다는 치명적인 문제가 있습니다. 창 전환 순간.
예를 들어, 이 예에서 00:02에 갑자기 많은 요청이 들어왔는데 이때 카운트를 재설정하면 갑작스러운 트래픽을 제한할 수 없습니다.
슬라이딩 윈도우 알고리즘
고정 윈도우의 갑작스러운 트래픽 문제를 완화하기 위해 슬라이딩 윈도우 알고리즘을 사용할 수 있습니다. 컴퓨터 네트워크의 TCP 흐름 제어에서는 슬라이딩 윈도우 알고리즘을 사용합니다.
알고리즘 원리
슬라이딩 창 전류 제한 알고리즘의 원리는 큰 시간 창을 여러 개의 작은 시간 창으로 나누고 각 작은 창은 독립적인 개수를 갖는 것입니다.
요청이 들어오면 요청 개수가 전체 창 제한을 초과하는지 확인하세요. 창은 매번 앞으로 이동합니다슬라이딩작은 단위 창. 滑动一个小的单元窗口。
例如下面这个滑动窗口，将大时间窗口1min分成了5个小窗口，每个小窗口的时间是12s。
每个单元格有自己独立的计数器，每过12s就会向前移动一格。
假如有请求在00:01的时候过来，这时候窗口的计数就是3+12+9+15=39，也能起到限流的作用。
这就是为什么滑动窗口能解决临界问题，滑的格子越多，那么整体的滑动就会越平滑
🎜슬라이딩 윈도우 알고리즘의 개략도🎜🎜🎜이것이 슬라이딩 윈도우가 존재하는 이유입니다 중요한 문제를 해결할 수 있습니다. 슬라이딩 그리드가 많을수록 전체 슬라이딩이 더 많아집니다. : rgba( 27, 31, 35, 0.05); 글꼴 계열: "Operator Mono", Consolas, Monaco, Menlo, monospace; 단어 나누기: break-all;color: rgb(255, 100, 65);"> Smooth code>, 전류 제한 효과가 더 정확해집니다. 🎜🎜알고리즘 구현🎜🎜그럼 여기서 슬라이딩 윈도우 전류 제한 알고리즘을 어떻게 구현합니까? 매우 간단합니다. Redis의 순서 집합(zset) 구조를 직접 사용할 수 있습니다. 🎜🎜타임스탬프를 점수와 멤버로 사용합니다. 요청이 오면 현재 타임스탬프가 주문한 세트에 추가됩니다. 그런 다음 창 외부의 요청에 대해 창 크기를 기준으로 시작 타임스탬프를 계산하고 창 외부의 요청을 삭제할 수 있습니다. 이런 방식으로 주문된 세트의 크기는 우리 창의 요청 수입니다. 🎜
代码实现
public class SlidingWindowRateLimiter {
public static final String KEY = "slidingWindowRateLimiter:";
/**
* 请求次数限制
*/
private Long limit;
/**
* 窗口大小（单位：S）
*/
private Long windowSize;
public SlidingWindowRateLimiter(Long limit, Long windowSize) {
this.limit = limit;
this.windowSize = windowSize;
}
public boolean triggerLimit(String path) {
RedissonClient redissonClient = RedissonConfig.getInstance();
//窗口计数
RScoredSortedSet<Long> counter = redissonClient.getScoredSortedSet(KEY + path);
//使用分布式锁，避免并发设置初始值的时候，导致窗口计数被覆盖
RLock rLock = redissonClient.getLock(KEY + "LOCK:" + path);
try {
rLock.lock(200, TimeUnit.MILLISECONDS);
// 当前时间戳
long currentTimestamp = System.currentTimeMillis();
// 窗口起始时间戳
long windowStartTimestamp = currentTimestamp - windowSize * 1000;
// 移除窗口外的时间戳，左闭右开
counter.removeRangeByScore(0, true, windowStartTimestamp, false);
// 将当前时间戳作为score,也作为member，
// TODO:高并发情况下可能没法保证唯一，可以加一个唯一标识
counter.add(currentTimestamp, currentTimestamp);
//使用zset的元素个数，作为请求计数
long count = counter.size();
// 判断时间戳数量是否超过限流阈值
if (count > limit) {
System.out.println("[triggerLimit] path:" + path + " count:" + count + " over limit:" + limit);
return true;
}
return false;
} finally {
rLock.unlock();
}
}
}这里还有一个小的可以完善的点，zset在member相同的情况下，是会覆盖的，也就是说高并发情况下，时间戳可能会重复，那么就有可能统计的请求偏少，这里可以用时间戳+随机数来缓解，也可以生成唯一序列来解决，比如UUID、雪花算法等等。
还是来测试一下
class SlidingWindowRateLimiterTest {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
@Test
@DisplayName("滑动窗口限流")
void triggerLimit() throws InterruptedException {
SlidingWindowRateLimiter slidingWindowRateLimiter = new SlidingWindowRateLimiter(10L, 1L);
//模拟在不同时间片内的请求
for (int i = 0; i < 8; i++) {
CountDownLatch countDownLatch = new CountDownLatch(20);
for (int j = 0; j < 20; j++) {
threadPoolExecutor.execute(() -> {
boolean isLimit = slidingWindowRateLimiter.triggerLimit("/test");
System.out.println(isLimit);
countDownLatch.countDown();
});
}
countDownLatch.await();
//休眠10s
TimeUnit.SECONDS.sleep(10L);
}
}
}用Redis实现了滑动窗口限流，解决了固定窗口限流的边界问题，当然这里也带来了新的问题，因为我们存储了窗口期的所有请求，所以高并发的情况下，可能会比较占内存。
漏桶算法
我们可以看到，计数器类的限流，体现的是一个“戛然而止”，超过限制，立马决绝，但是有时候，我们可能只是希望请求平滑一些，追求的是“波澜不惊”，这时候就可以考虑使用其它的限流算法。
算法原理
漏桶算法（Leaky Bucket），名副其实，就是请求就像水一样以任意速度注入漏桶，而桶会按照固定的速率将水漏掉。
当进水速率大于出水速率的时候，漏桶会变满，此时新进入的请求将会被丢弃。
漏桶算法的两大作用是网络流量整形（Traffic Shaping）和速度限制（Rate Limiting）。
算法实现
我们接着看看具体应该怎么实现。
在滑动窗口限流算法里我们用到了RScoredSortedSet，非常好用对不对，这里也可以用这个结构，直接使用ZREMRANGEBYSCORE命令来删除旧的请求。
进水就不用多说了，请求进来，判断桶有没有满，满了就拒绝，没满就往桶里丢请求。
那么出水怎么办呢？得保证稳定速率出水，可以用一个定时任务，来定时去删除旧的请求。
代码实现
public class LeakyBucketRateLimiter {
private RedissonClient redissonClient = RedissonConfig.getInstance();
private static final String KEY_PREFIX = "LeakyBucket:";
/**
* 桶的大小
*/
private Long bucketSize;
/**
* 漏水速率，单位:个/秒
*/
private Long leakRate;
public LeakyBucketRateLimiter(Long bucketSize, Long leakRate) {
this.bucketSize = bucketSize;
this.leakRate = leakRate;
//这里启动一个定时任务，每s执行一次
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
executorService.scheduleAtFixedRate(this::leakWater, 0, 1, TimeUnit.SECONDS);
}
/**
* 漏水
*/
public void leakWater() {
RSet<String> pathSet=redissonClient.getSet(KEY_PREFIX+":pathSet");
//遍历所有path,删除旧请求
for(String path:pathSet){
String redisKey = KEY_PREFIX + path;
RScoredSortedSet<Long> bucket = redissonClient.getScoredSortedSet(KEY_PREFIX + path);
// 获取当前时间
long now = System.currentTimeMillis();
// 删除旧的请求
bucket.removeRangeByScore(0, true,now - 1000 * leakRate,true);
}
}
/**
* 限流
*/
public boolean triggerLimit(String path) {
//加锁，防止并发初始化问题
RLock rLock = redissonClient.getLock(KEY_PREFIX + "LOCK:" + path);
try {
rLock.lock(100,TimeUnit.MILLISECONDS);
String redisKey = KEY_PREFIX + path;
RScoredSortedSet<Long> bucket = redissonClient.getScoredSortedSet(redisKey);
//这里用一个set，来存储所有path
RSet<String> pathSet=redissonClient.getSet(KEY_PREFIX+":pathSet");
pathSet.add(path);
// 获取当前时间
long now = System.currentTimeMillis();
// 检查桶是否已满
if (bucket.size() < bucketSize) {
// 桶未满，添加一个元素到桶中
bucket.add(now,now);
return false;
}
// 桶已满，触发限流
System.out.println("[triggerLimit] path:"+path+" bucket size:"+bucket.size());
return true;
}finally {
rLock.unlock();
}
}
}在代码实现里，我们用了RSet来存储path，这样一来，一个定时任务，就可以搞定所有path对应的桶的出水，而不用每个桶都创建一个一个定时任务。
这里我直接用ScheduledExecutorService启动了一个定时任务，1s跑一次，当然集群环境下，每台机器都跑一个定时任务，对性能是极大的浪费，而且不好管理，我们可以用分布式定时任务，比如xxl-job去执行leakWater。
最后还是大家熟悉的测试
class LeakyBucketRateLimiterTest {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
@Test
@DisplayName("漏桶算法")
void triggerLimit() throws InterruptedException {
LeakyBucketRateLimiter leakyBucketRateLimiter = new LeakyBucketRateLimiter(10L, 1L);
for (int i = 0; i < 8; i++) {
CountDownLatch countDownLatch = new CountDownLatch(20);
for (int j = 0; j < 20; j++) {
threadPoolExecutor.execute(() -> {
boolean isLimit = leakyBucketRateLimiter.triggerLimit("/test");
System.out.println(isLimit);
countDownLatch.countDown();
});
}
countDownLatch.await();
//休眠10s
TimeUnit.SECONDS.sleep(10L);
}
}
}漏桶算法能够有效防止网络拥塞，实现也比较简单。
但是，因为漏桶的出水速率是固定的，假如突然来了大量的请求，那么只能丢弃超量的请求，即使下游能处理更大的流量，没法充分利用系统资源。
令牌桶算法
令牌桶算法来了！
算法原理
令牌桶算法是对漏桶算法的一种改进。
它的主要思想是：系统以一种固定的速率向桶中添加令牌，每个请求在发送前都需要从桶中取出一个令牌，只有取到令牌的请求才被通过。因此，令牌桶算法允许请求以任意速率发送，只要桶中有足够的令牌。
算法实现
我们继续看怎么实现，首先是要发放令牌，要固定速率，那我们又得开个线程，定时往桶里投令牌，然后……
——然后Redission提供了令牌桶算法的实现，舒不舒服？
拿来就用！
代码实现
public class TokenBucketRateLimiter {
public static final String KEY = "TokenBucketRateLimiter:";
/**
* 阈值
*/
private Long limit;
/**
* 添加令牌的速率，单位：个/秒
*/
private Long tokenRate;
public TokenBucketRateLimiter(Long limit, Long tokenRate) {
this.limit = limit;
this.tokenRate = tokenRate;
}
/**
* 限流算法
*/
public boolean triggerLimit(String path){
RedissonClient redissonClient=RedissonConfig.getInstance();
RRateLimiter rateLimiter = redissonClient.getRateLimiter(KEY+path);
// 初始化，设置速率模式，速率，间隔，间隔单位
rateLimiter.trySetRate(RateType.OVERALL, limit, tokenRate, RateIntervalUnit.SECONDS);
// 获取令牌
return rateLimiter.tryAcquire();
}
}Redisson实现的，还是比较稳的，这里就不测试了。
关于Redission是怎么实现这个限速器的，大家可以看一下参考[3]，还是Redisson家的老传统——Lua脚本，设计相当巧妙。
总结
在这篇文章里，我们对四（三）种限流算法进行了分布式实现，采用了非常好用的Redission客户端，当然我们也有不完善的地方：
并发处理采用了分布式锁，高并发情况下，对性能有一定损耗，逻辑最好还是直接采用Lua脚本实现，来提高性能 可以提供更加优雅的调用方式，比如利用aop实现注解式调用，代码设计也可以更加优雅，继承体系可以完善一下 没有实现限流的拒绝策略，比如抛异常、缓存、丢进MQ打散……限流是一种方法，最终的目的还是尽可能保证系统平稳
如果后面有机会，希望可以继续完善这个简单的Demo，达到工程级的应用。
除此之外，市面上也有很多好用的开源限流工具：
Guava RateLimiter ，基于令牌桶算法限流，当然是单机的； Sentinel ，基于滑动窗口限流，支持单机，也支持集群 网关限流，很多网关自带限流方法，比如 Spring Cloud Gateway、Nginx
위 내용은 4개의 분산 전류 제한 알고리즘 및 코드 구현의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!