Skip to content

Commit 70b3983

Browse files
dunwudunwu
committed
feat: 限流算法示例
1 parent aad3a98 commit 70b3983

8 files changed

Lines changed: 415 additions & 0 deletions

File tree

‎codes/java-distributed/java-rate-limit/pom.xml‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
3+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4+
<modelVersion>4.0.0</modelVersion>
5+
6+
<groupId>io.github.dunwu.javatech</groupId>
7+
<artifactId>java-rate-limit</artifactId>
8+
<version>1.0.0</version>
9+
<packaging>jar</packaging>
10+
<name>${project.artifactId}</name>
11+
12+
<properties>
13+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
14+
<java.version>1.8</java.version>
15+
<maven.compiler.source>${java.version}</maven.compiler.source>
16+
<maven.compiler.target>${java.version}</maven.compiler.target>
17+
</properties>
18+
19+
<dependencies>
20+
<dependency>
21+
<groupId>cn.hutool</groupId>
22+
<artifactId>hutool-all</artifactId>
23+
<version>5.8.25</version>
24+
</dependency>
25+
<dependency>
26+
<groupId>org.projectlombok</groupId>
27+
<artifactId>lombok</artifactId>
28+
<version>1.18.30</version>
29+
</dependency>
30+
<dependency>
31+
<groupId>ch.qos.logback</groupId>
32+
<artifactId>logback-classic</artifactId>
33+
<version>1.2.3</version>
34+
<optional>true</optional>
35+
</dependency>
36+
</dependencies>
37+
</project>

‎codes/java-distributed/java-rate-limit/src/main/java/io/github/dunwu/distributed/ratelimit/FixedWindowRateLimiter.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package io.github.dunwu.distributed.ratelimit;
2+
3+
import java.util.concurrent.TimeUnit;
4+
import java.util.concurrent.atomic.AtomicLong;
5+
6+
/**
7+
* 固定时间窗口限流算法
8+
*
9+
* @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
10+
* @date 2024-01-18
11+
*/
12+
public class FixedWindowRateLimiter implements RateLimiter {
13+
14+
/**
15+
* 允许的最大请求数
16+
*/
17+
private final long maxPermits;
18+
19+
/**
20+
* 窗口期时长
21+
*/
22+
private final long periodMillis;
23+
24+
/**
25+
* 窗口期截止时间
26+
*/
27+
private long lastPeriodMillis;
28+
29+
/**
30+
* 请求计数
31+
*/
32+
private AtomicLong count = new AtomicLong(0);
33+
34+
public FixedWindowRateLimiter(long qps) {
35+
this(qps, 1000, TimeUnit.MILLISECONDS);
36+
}
37+
38+
public FixedWindowRateLimiter(long maxPermits, long period, TimeUnit timeUnit) {
39+
this.maxPermits = maxPermits;
40+
this.periodMillis = timeUnit.toMillis(period);
41+
this.lastPeriodMillis = System.currentTimeMillis() + this.periodMillis;
42+
}
43+
44+
@Override
45+
public synchronized boolean tryAcquire(int permits) {
46+
long now = System.currentTimeMillis();
47+
if (lastPeriodMillis <= now) {
48+
this.lastPeriodMillis = now + this.periodMillis;
49+
count = new AtomicLong(0);
50+
}
51+
if (count.get() + permits <= maxPermits) {
52+
count.addAndGet(permits);
53+
return true;
54+
} else {
55+
return false;
56+
}
57+
}
58+
59+
}

‎codes/java-distributed/java-rate-limit/src/main/java/io/github/dunwu/distributed/ratelimit/LeakyBucketRateLimiter.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
package io.github.dunwu.distributed.ratelimit;
2+
3+
import java.util.concurrent.atomic.AtomicLong;
4+
5+
/**
6+
* 漏桶限流算法
7+
*
8+
* @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
9+
* @date 2024-01-18
10+
*/
11+
public class LeakyBucketRateLimiter implements RateLimiter {
12+
13+
/**
14+
* QPS
15+
*/
16+
private final int qps;
17+
18+
/**
19+
* 桶的容量
20+
*/
21+
private final long capacity;
22+
23+
/**
24+
* 计算的起始时间
25+
*/
26+
private long beginTimeMillis;
27+
28+
/**
29+
* 桶中当前的水量
30+
*/
31+
private final AtomicLong waterNum = new AtomicLong(0);
32+
33+
public LeakyBucketRateLimiter(int qps, int capacity) {
34+
this.qps = qps;
35+
this.capacity = capacity;
36+
}
37+
38+
@Override
39+
public synchronized boolean tryAcquire(int permits) {
40+
41+
// 如果桶中没有水，直接放行
42+
if (waterNum.get() == 0) {
43+
beginTimeMillis = System.currentTimeMillis();
44+
waterNum.addAndGet(permits);
45+
return true;
46+
}
47+
48+
// 计算水量
49+
long leakedWaterNum = ((System.currentTimeMillis() - beginTimeMillis) / 1000) * qps;
50+
long currentWaterNum = waterNum.get() - leakedWaterNum;
51+
waterNum.set(Math.max(0, currentWaterNum));
52+
53+
// 重置时间
54+
beginTimeMillis = System.currentTimeMillis();
55+
56+
if (waterNum.get() + permits < capacity) {
57+
waterNum.addAndGet(permits);
58+
return true;
59+
} else {
60+
return false;
61+
}
62+
}
63+
64+
}

‎codes/java-distributed/java-rate-limit/src/main/java/io/github/dunwu/distributed/ratelimit/RateLimiter.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
package io.github.dunwu.distributed.ratelimit;
2+
3+
/**
4+
* 限流器
5+
*
6+
* @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
7+
* @date 2024-01-18
8+
*/
9+
public interface RateLimiter {
10+
11+
boolean tryAcquire(int permits);
12+
13+
}

‎codes/java-distributed/java-rate-limit/src/main/java/io/github/dunwu/distributed/ratelimit/RateLimiterDemo.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
package io.github.dunwu.distributed.ratelimit;
2+
3+
import cn.hutool.core.thread.ThreadUtil;
4+
import cn.hutool.core.util.RandomUtil;
5+
import lombok.extern.slf4j.Slf4j;
6+
7+
import java.util.concurrent.CountDownLatch;
8+
import java.util.concurrent.ExecutorService;
9+
import java.util.concurrent.TimeUnit;
10+
import java.util.concurrent.atomic.AtomicInteger;
11+
12+
/**
13+
* 限流器示例
14+
*
15+
* @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
16+
* @date 2024-01-18
17+
*/
18+
@Slf4j
19+
public class RateLimiterDemo {
20+
21+
public static void main(String[] args) {
22+
23+
// ============================================================================
24+
25+
int qps = 20;
26+
27+
System.out.println("======================= 固定时间窗口限流算法 =======================");
28+
FixedWindowRateLimiter fixedWindowRateLimiter = new FixedWindowRateLimiter(qps);
29+
testRateLimit(fixedWindowRateLimiter, qps);
30+
31+
System.out.println("======================= 滑动时间窗口限流算法 =======================");
32+
SlidingWindowRateLimiter slidingWindowRateLimiter = new SlidingWindowRateLimiter(qps, 10);
33+
testRateLimit(slidingWindowRateLimiter, qps);
34+
35+
System.out.println("======================= 漏桶限流算法 =======================");
36+
LeakyBucketRateLimiter leakyBucketRateLimiter = new LeakyBucketRateLimiter(qps, 100);
37+
testRateLimit(leakyBucketRateLimiter, qps);
38+
39+
System.out.println("======================= 令牌桶限流算法 =======================");
40+
TokenBucketRateLimiter tokenBucketRateLimiter = new TokenBucketRateLimiter(qps, 100);
41+
testRateLimit(tokenBucketRateLimiter, qps);
42+
}
43+
44+
private static void testRateLimit(RateLimiter rateLimiter, int qps) {
45+
46+
AtomicInteger okNum = new AtomicInteger(0);
47+
AtomicInteger limitNum = new AtomicInteger(0);
48+
ExecutorService executorService = ThreadUtil.newFixedExecutor(10, "限流测试", true);
49+
long beginTime = System.currentTimeMillis();
50+
51+
int threadNum = 4;
52+
final CountDownLatch latch = new CountDownLatch(threadNum);
53+
for (int i = 0; i < threadNum; i++) {
54+
executorService.submit(() -> {
55+
try {
56+
batchRequest(rateLimiter, okNum, limitNum, 1000);
57+
} catch (Exception e) {
58+
log.error("发生异常!", e);
59+
} finally {
60+
latch.countDown();
61+
}
62+
});
63+
}
64+
65+
try {
66+
latch.await(10, TimeUnit.SECONDS);
67+
long endTime = System.currentTimeMillis();
68+
long gap = endTime - beginTime;
69+
log.info("限流 QPS: {} -> 实际结果：耗时 {} ms，{} 次请求成功，{} 次请求被限流，实际 QPS: {}",
70+
qps, gap, okNum.get(), limitNum.get(), okNum.get() * 1000 / gap);
71+
if (okNum.get() == qps) {
72+
log.info("限流符合预期");
73+
}
74+
} catch (Exception e) {
75+
log.error("发生异常!", e);
76+
} finally {
77+
executorService.shutdown();
78+
}
79+
}
80+
81+
private static void batchRequest(RateLimiter rateLimiter, AtomicInteger okNum, AtomicInteger limitNum, int num)
82+
throws InterruptedException {
83+
for (int j = 0; j < num; j++) {
84+
if (rateLimiter.tryAcquire(1)) {
85+
log.info("请求成功");
86+
okNum.getAndIncrement();
87+
} else {
88+
log.info("请求限流");
89+
limitNum.getAndIncrement();
90+
}
91+
TimeUnit.MILLISECONDS.sleep(RandomUtil.randomInt(0, 10));
92+
}
93+
}
94+
95+
}

‎codes/java-distributed/java-rate-limit/src/main/java/io/github/dunwu/distributed/ratelimit/SlidingWindowRateLimiter.java‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
package io.github.dunwu.distributed.ratelimit;
2+
3+
import java.util.LinkedList;
4+
import java.util.List;
5+
import java.util.concurrent.TimeUnit;
6+
import java.util.concurrent.atomic.AtomicLong;
7+
8+
/**
9+
* 滑动时间窗口限流算法
10+
*
11+
* @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
12+
* @date 2024-01-18
13+
*/
14+
public class SlidingWindowRateLimiter implements RateLimiter {
15+
16+
/**
17+
* 允许的最大请求数
18+
*/
19+
private final long maxPermits;
20+
21+
/**
22+
* 窗口期时长
23+
*/
24+
private final long periodMillis;
25+
26+
/**
27+
* 分片窗口期时长
28+
*/
29+
private final long shardPeriodMillis;
30+
31+
/**
32+
* 窗口期截止时间
33+
*/
34+
private long lastPeriodMillis;
35+
36+
/**
37+
* 分片窗口数
38+
*/
39+
private final int shardNum;
40+
41+
/**
42+
* 请求总计数
43+
*/
44+
private final AtomicLong totalCount = new AtomicLong(0);
45+
46+
/**
47+
* 分片窗口计数列表
48+
*/
49+
private final List<AtomicLong> countList = new LinkedList<>();
50+
51+
public SlidingWindowRateLimiter(long qps, int shardNum) {
52+
this(qps, 1000, TimeUnit.MILLISECONDS, shardNum);
53+
}
54+
55+
public SlidingWindowRateLimiter(long maxPermits, long period, TimeUnit timeUnit, int shardNum) {
56+
this.maxPermits = maxPermits;
57+
this.periodMillis = timeUnit.toMillis(period);
58+
this.lastPeriodMillis = System.currentTimeMillis();
59+
this.shardPeriodMillis = timeUnit.toMillis(period) / shardNum;
60+
this.shardNum = shardNum;
61+
for (int i = 0; i < shardNum; i++) {
62+
countList.add(new AtomicLong(0));
63+
}
64+
}
65+
66+
@Override
67+
public synchronized boolean tryAcquire(int permits) {
68+
long now = System.currentTimeMillis();
69+
if (now > lastPeriodMillis) {
70+
for (int shardId = 0; shardId < shardNum; shardId++) {
71+
long shardCount = countList.get(shardId).get();
72+
totalCount.addAndGet(-shardCount);
73+
countList.set(shardId, new AtomicLong(0));
74+
lastPeriodMillis += shardPeriodMillis;
75+
}
76+
}
77+
int shardId = (int) (now % periodMillis / shardPeriodMillis);
78+
if (totalCount.get() + permits <= maxPermits) {
79+
countList.get(shardId).addAndGet(permits);
80+
totalCount.addAndGet(permits);
81+
return true;
82+
} else {
83+
return false;
84+
}
85+
}
86+
87+
}

0 commit comments

Comments
 (0)