mirror of
https://github.com/tiennm99/java-design-patterns.git
synced 2026-10-12 22:14:41 +00:00
fix: Fix test stability issues
This commit is contained in:
1 parent
13cde46f2e
commit
177dc52704
24 files changed
+253
-111
No files matched your search
+33
-4
@@ -10,13 +10,28 @@ public class AdaptiveRateLimiter implements RateLimiter {
|
||||
private final AtomicInteger currentLimit;
|
||||
private final ConcurrentHashMap<String, RateLimiter> limiters = new ConcurrentHashMap<>();
|
||||
private final ScheduledExecutorService healthChecker = Executors.newScheduledThreadPool(1);
|
||||
private final long checkInterval;
|
||||
private final TimeUnit timeUnit;
|
||||
|
||||
public AdaptiveRateLimiter(int initialLimit, int maxLimit) {
|
||||
this(initialLimit, maxLimit, 10, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructor with option to set custom interval.
|
||||
*
|
||||
* @param initialLimit initial rate limit
|
||||
* @param maxLimit maximum rate limit
|
||||
* @param interval check interval
|
||||
* @param timeUnit check interval time unit
|
||||
*/
|
||||
public AdaptiveRateLimiter(int initialLimit, int maxLimit, long interval, TimeUnit timeUnit) {
|
||||
this.initialLimit = initialLimit;
|
||||
this.maxLimit = maxLimit;
|
||||
this.currentLimit = new AtomicInteger(initialLimit);
|
||||
// Periodically increase limit to recover if system appears healthy
|
||||
healthChecker.scheduleAtFixedRate(this::adjustLimits, 10, 10, TimeUnit.SECONDS);
|
||||
this.checkInterval = interval;
|
||||
this.timeUnit = timeUnit;
|
||||
healthChecker.scheduleAtFixedRate(this::adjustLimits, interval, interval, timeUnit);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -24,9 +39,10 @@ public class AdaptiveRateLimiter implements RateLimiter {
|
||||
String key = serviceName + ":" + operationName;
|
||||
int current = currentLimit.get();
|
||||
|
||||
// Reuse or create TokenBucket for this key using currentLimit
|
||||
// Reuse or create TokenBucket for this key using currentLimit and checkInterval
|
||||
RateLimiter limiter =
|
||||
limiters.computeIfAbsent(key, k -> new TokenBucketRateLimiter(current, current));
|
||||
limiters.computeIfAbsent(
|
||||
key, k -> new TokenBucketRateLimiter(current, current, checkInterval, timeUnit));
|
||||
|
||||
try {
|
||||
limiter.check(serviceName, operationName);
|
||||
@@ -47,4 +63,17 @@ public class AdaptiveRateLimiter implements RateLimiter {
|
||||
int updated = currentLimit.updateAndGet(curr -> Math.min(maxLimit, curr + (initialLimit / 2)));
|
||||
System.out.printf("[Adaptive] Health check passed - Increasing limit to %d%n", updated);
|
||||
}
|
||||
|
||||
/** Gracefully shut down the health checker executor and child limiters. */
|
||||
public void shutdown() {
|
||||
healthChecker.shutdown();
|
||||
limiters
|
||||
.values()
|
||||
.forEach(
|
||||
limiter -> {
|
||||
if (limiter instanceof TokenBucketRateLimiter) {
|
||||
((TokenBucketRateLimiter) limiter).shutdown();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
+30
-7
@@ -4,8 +4,10 @@ import java.util.concurrent.*;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* Token Bucket rate limiter implementation. Allows requests to proceed as long as there are tokens
|
||||
* available in the bucket. Tokens are added at a fixed interval up to a defined capacity.
|
||||
* Token Bucket rate limiter implementation. Allows requests to proceed as long
|
||||
* as there are tokens
|
||||
* available in the bucket. Tokens are added at a fixed interval up to a defined
|
||||
* capacity.
|
||||
*/
|
||||
public class TokenBucketRateLimiter implements RateLimiter {
|
||||
private final int capacity;
|
||||
@@ -14,10 +16,22 @@ public class TokenBucketRateLimiter implements RateLimiter {
|
||||
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
|
||||
|
||||
public TokenBucketRateLimiter(int capacity, int refillRate) {
|
||||
this(capacity, refillRate, 1, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructor with custom refill interval.
|
||||
*
|
||||
* @param capacity token bucket capacity
|
||||
* @param refillRate token refill rate
|
||||
* @param refillInterval refill interval value
|
||||
* @param timeUnit refill interval time unit
|
||||
*/
|
||||
public TokenBucketRateLimiter(
|
||||
int capacity, int refillRate, long refillInterval, TimeUnit timeUnit) {
|
||||
this.capacity = capacity;
|
||||
this.refillRate = refillRate;
|
||||
// Refill tokens in all buckets every second
|
||||
scheduler.scheduleAtFixedRate(this::refillBuckets, 1, 1, TimeUnit.SECONDS);
|
||||
scheduler.scheduleAtFixedRate(this::refillBuckets, refillInterval, refillInterval, timeUnit);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -39,7 +53,10 @@ public class TokenBucketRateLimiter implements RateLimiter {
|
||||
buckets.forEach((k, b) -> b.refill(refillRate));
|
||||
}
|
||||
|
||||
/** Inner class that represents the bucket holding tokens for each service-operation. */
|
||||
/**
|
||||
* Inner class that represents the bucket holding tokens for each
|
||||
* service-operation.
|
||||
*/
|
||||
private static class TokenBucket {
|
||||
private final int capacity;
|
||||
private final AtomicInteger tokens;
|
||||
@@ -52,8 +69,10 @@ public class TokenBucketRateLimiter implements RateLimiter {
|
||||
boolean tryConsume() {
|
||||
while (true) {
|
||||
int current = tokens.get();
|
||||
if (current <= 0) return false;
|
||||
if (tokens.compareAndSet(current, current - 1)) return true;
|
||||
if (current <= 0)
|
||||
return false;
|
||||
if (tokens.compareAndSet(current, current - 1))
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -61,4 +80,8 @@ public class TokenBucketRateLimiter implements RateLimiter {
|
||||
tokens.getAndUpdate(current -> Math.min(current + amount, capacity));
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
scheduler.shutdown();
|
||||
}
|
||||
}
|
||||
+46
-35
@@ -8,49 +8,60 @@ class AdaptiveRateLimiterTest {
|
||||
@Test
|
||||
void shouldDecreaseLimitWhenThrottled() throws Exception {
|
||||
AdaptiveRateLimiter limiter = new AdaptiveRateLimiter(10, 20);
|
||||
|
||||
// Exceed initial limit
|
||||
for (int i = 0; i < 11; i++) {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
} catch (RateLimitException e) {
|
||||
// Expected after 10 requests
|
||||
try {
|
||||
// Exceed initial limit
|
||||
for (int i = 0; i < 11; i++) {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
} catch (RateLimitException e) {
|
||||
// Expected after 10 requests
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Verify limit was reduced
|
||||
assertThrows(
|
||||
RateLimitException.class,
|
||||
() -> {
|
||||
for (int i = 0; i < 6; i++) { // New limit should be 5 (10/2)
|
||||
limiter.check("test", "op");
|
||||
}
|
||||
});
|
||||
// Verify limit was reduced
|
||||
assertThrows(
|
||||
RateLimitException.class,
|
||||
() -> {
|
||||
for (int i = 0; i < 6; i++) { // New limit should be 5 (10/2)
|
||||
limiter.check("test", "op");
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
limiter.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldGraduallyIncreaseLimitWhenHealthy() throws Exception {
|
||||
AdaptiveRateLimiter limiter =
|
||||
new AdaptiveRateLimiter(4, 10); // Start from 4 → expect 2 → expect increase to 4
|
||||
|
||||
// Force throttling to reduce limit
|
||||
for (int i = 0; i < 5; i++) {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
} catch (RateLimitException e) {
|
||||
// Expected to throttle and reduce limit
|
||||
new AdaptiveRateLimiter(
|
||||
4,
|
||||
10,
|
||||
100,
|
||||
java.util.concurrent.TimeUnit
|
||||
.MILLISECONDS); // Start from 4 → expect 2 → expect increase to 4
|
||||
try {
|
||||
// Force throttling to reduce limit
|
||||
for (int i = 0; i < 5; i++) {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
} catch (RateLimitException e) {
|
||||
// Expected to throttle and reduce limit
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for health check to increase limit
|
||||
Thread.sleep(150); // Wait slightly more than 100 milliseconds
|
||||
|
||||
// Allow up to 4 requests again (limit should've increased to 4)
|
||||
for (int i = 0; i < 4; i++) {
|
||||
limiter.check("test", "op");
|
||||
}
|
||||
|
||||
// 5th should throw exception again
|
||||
assertThrows(RateLimitException.class, () -> limiter.check("test", "op"));
|
||||
} finally {
|
||||
limiter.shutdown();
|
||||
}
|
||||
|
||||
// Wait for health check to increase limit
|
||||
Thread.sleep(11000); // Wait slightly more than 10 seconds
|
||||
|
||||
// Allow up to 4 requests again (limit should've increased to 4)
|
||||
for (int i = 0; i < 4; i++) {
|
||||
limiter.check("test", "op");
|
||||
}
|
||||
|
||||
// 5th should throw exception again
|
||||
assertThrows(RateLimitException.class, () -> limiter.check("test", "op"));
|
||||
}
|
||||
}
|
||||
+50
-42
@@ -13,57 +13,65 @@ class ConcurrencyTests {
|
||||
int requestLimit = 5;
|
||||
RateLimiter limiter = new TokenBucketRateLimiter(requestLimit, requestLimit);
|
||||
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
|
||||
CountDownLatch latch = new CountDownLatch(threadCount);
|
||||
try {
|
||||
CountDownLatch latch = new CountDownLatch(threadCount);
|
||||
|
||||
AtomicInteger successCount = new AtomicInteger();
|
||||
AtomicInteger failureCount = new AtomicInteger();
|
||||
AtomicInteger successCount = new AtomicInteger();
|
||||
AtomicInteger failureCount = new AtomicInteger();
|
||||
|
||||
for (int i = 0; i < threadCount; i++) {
|
||||
executor.submit(
|
||||
() -> {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
successCount.incrementAndGet();
|
||||
} catch (RateLimitException e) {
|
||||
failureCount.incrementAndGet();
|
||||
}
|
||||
latch.countDown();
|
||||
});
|
||||
for (int i = 0; i < threadCount; i++) {
|
||||
executor.submit(
|
||||
() -> {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
successCount.incrementAndGet();
|
||||
} catch (RateLimitException e) {
|
||||
failureCount.incrementAndGet();
|
||||
}
|
||||
latch.countDown();
|
||||
});
|
||||
}
|
||||
|
||||
latch.await();
|
||||
assertEquals(requestLimit, successCount.get());
|
||||
assertEquals(threadCount - requestLimit, failureCount.get());
|
||||
} finally {
|
||||
executor.shutdown();
|
||||
}
|
||||
|
||||
latch.await();
|
||||
assertEquals(requestLimit, successCount.get());
|
||||
assertEquals(threadCount - requestLimit, failureCount.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void adaptiveLimiterShouldAdjustUnderLoad() throws Exception {
|
||||
AdaptiveRateLimiter limiter = new AdaptiveRateLimiter(10, 20);
|
||||
AdaptiveRateLimiter limiter = new AdaptiveRateLimiter(10, 20, 100, TimeUnit.MILLISECONDS);
|
||||
ExecutorService executor = Executors.newFixedThreadPool(20);
|
||||
|
||||
// Flood with requests to trigger throttling
|
||||
for (int i = 0; i < 30; i++) {
|
||||
executor.submit(
|
||||
() -> {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
} catch (RateLimitException ignored) {
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Thread.sleep(15000); // Wait for adjustment
|
||||
|
||||
// Verify new limit is in effect
|
||||
int allowed = 0;
|
||||
for (int i = 0; i < 20; i++) {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
allowed++;
|
||||
} catch (RateLimitException ignored) {
|
||||
try {
|
||||
// Flood with requests to trigger throttling
|
||||
for (int i = 0; i < 30; i++) {
|
||||
executor.submit(
|
||||
() -> {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
} catch (RateLimitException ignored) {
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
assertTrue(allowed > 5 && allowed < 15); // Should be between initial and max
|
||||
Thread.sleep(150); // Wait for adjustment
|
||||
|
||||
// Verify new limit is in effect
|
||||
int allowed = 0;
|
||||
for (int i = 0; i < 20; i++) {
|
||||
try {
|
||||
limiter.check("test", "op");
|
||||
allowed++;
|
||||
} catch (RateLimitException ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
assertTrue(allowed > 5 && allowed < 15); // Should be between initial and max
|
||||
} finally {
|
||||
executor.shutdown();
|
||||
limiter.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user