iSharkFly-Docs/java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

BAEL-5764 - Limiting the Request per Second With Webclient (#12837)

Browse Source 
* bael-5764 - ready for review

* bael-5764 - editor review 1

* bael-5764 - editor review 2
This commit is contained in:
Ulisses Lima 2022-10-27 07:34:20 -03:00 committed by GitHub
parent 7bc0532d95
commit 8e709cd717
18 changed files with 593 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
spring-reactive-modules/pom.xml
View File

@ -22,6 +22,7 @@
<module>spring-5-reactive-2</module>
<module>spring-5-reactive-3</module>
<module>spring-5-reactive-client</module>
<module>spring-5-reactive-client-2</module>
<module>spring-5-reactive-oauth</module>
<module>spring-5-reactive-security</module>
<module>spring-reactive</module>
@ -61,4 +62,4 @@
<properties>
</properties>
</project>
</project>

9
spring-reactive-modules/spring-5-reactive-client-2/README.md Normal file
View File

@ -0,0 +1,9 @@
## Spring REST Example Project
This module contains articles about reactive Spring 5 WebClient
### The Course
The "REST With Spring" Classes: http://bit.ly/restwithspring
### Relevant Articles
- More articles: [[<-- prev]](../spring-5-reactive-client)

146
spring-reactive-modules/spring-5-reactive-client-2/pom.xml Normal file
View File

@ -0,0 +1,146 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-5-reactive-client-2</artifactId>
<name>spring-5-reactive-client-2</name>
<packaging>jar</packaging>
<description>spring 5 sample project about new features</description>
<parent>
<groupId>com.baeldung.spring.reactive</groupId>
<artifactId>spring-reactive-modules</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.projectreactor</groupId>
<artifactId>reactor-spring</artifactId>
<version>${reactor-spring.version}</version>
</dependency>
<dependency>
<groupId>javax.json.bind</groupId>
<artifactId>javax.json.bind-api</artifactId>
</dependency>
<dependency>
<groupId>io.github.resilience4j</groupId>
<artifactId>resilience4j-reactor</artifactId>
<version>${resilience4j.version}</version>
</dependency>
<dependency>
<groupId>io.github.resilience4j</groupId>
<artifactId>resilience4j-ratelimiter</artifactId>
<version>${resilience4j.version}</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>${guava.version}</version>
</dependency>
<!-- runtime and test scoped -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-reactive-httpclient</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>integration-lite-first</id>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemPropertyVariables>
<logback.configurationFile>${project.basedir}/src/test/resources/logback-test.xml</logback.configurationFile>
</systemPropertyVariables>
</configuration>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>integration-lite-second</id>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemPropertyVariables>
<logback.configurationFile>${project.basedir}/src/test/resources/logback-test.xml</logback.configurationFile>
</systemPropertyVariables>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
<properties>
<reactor-spring.version>1.0.1.RELEASE</reactor-spring.version>
<jetty-reactive-httpclient.version>1.1.6</jetty-reactive-httpclient.version>
<reactor-test.version>3.2.10.RELEASE</reactor-test.version>
<resilience4j.version>1.7.1</resilience4j.version>
</properties>
</project>

12
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/LimitRequestsApp.java Normal file
View File

@ -0,0 +1,12 @@
package com.baeldung.limitrequests;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class LimitRequestsApp {
public static void main(String[] args) {
SpringApplication.run(LimitRequestsApp.class, args);
}
}

33
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/DelayElements.java Normal file
View File

@ -0,0 +1,33 @@
package com.baeldung.limitrequests.client;
import java.time.Duration;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.limitrequests.client.utils.Client;
import com.baeldung.limitrequests.client.utils.RandomConsumer;
import reactor.core.publisher.Flux;
public class DelayElements {
private DelayElements() {
}
public static Flux<Integer> fetch(WebClient client, int requests, int delay) {
String clientId = Client.id(requests, DelayElements.class, delay);
return Flux.range(1, requests)
.log()
.delayElements(Duration.ofMillis(delay))
.flatMap(i -> RandomConsumer.get(client, clientId));
}
public static void main(String[] args) {
String baseUrl = args[0];
WebClient client = WebClient.create(baseUrl);
fetch(client, 12, 1050).doOnNext(System.out::println)
.blockLast();
}
}

37
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/GuavaRateLimit.java Normal file
View File

@ -0,0 +1,37 @@
package com.baeldung.limitrequests.client;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.limitrequests.client.utils.Client;
import com.baeldung.limitrequests.client.utils.RandomConsumer;
import com.google.common.util.concurrent.RateLimiter;
import reactor.core.publisher.Flux;
public class GuavaRateLimit {
private GuavaRateLimit() {
}
public static Flux<Integer> fetch(WebClient client, int requests, double limit) {
String clientId = Client.id(requests, GuavaRateLimit.class, limit);
RateLimiter limiter = RateLimiter.create(limit);
return Flux.range(1, requests)
.log()
.flatMap(i -> {
limiter.acquire();
return RandomConsumer.get(client, clientId);
});
}
public static void main(String[] args) throws InterruptedException {
String baseUrl = args[0];
WebClient client = WebClient.create(baseUrl);
fetch(client, 20, 2).doOnNext(System.out::println)
.blockLast();
}
}

30
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/LimitConcurrency.java Normal file
View File

@ -0,0 +1,30 @@
package com.baeldung.limitrequests.client;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.limitrequests.client.utils.Client;
import com.baeldung.limitrequests.client.utils.RandomConsumer;
import reactor.core.publisher.Flux;
public class LimitConcurrency {
private LimitConcurrency() {
}
public static Flux<Integer> fetch(WebClient client, int requests, int concurrency) {
String clientId = Client.id(requests, LimitConcurrency.class.getSimpleName(), concurrency);
return Flux.range(1, requests)
.log()
.flatMap(i -> RandomConsumer.get(client, clientId), concurrency);
}
public static void main(String[] args) throws InterruptedException {
String baseUrl = args[0];
WebClient client = WebClient.create(baseUrl);
fetch(client, 12, 5).doOnNext(System.out::println)
.blockLast();
}
}

42
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/Resilience4jRateLimit.java Normal file
View File

@ -0,0 +1,42 @@
package com.baeldung.limitrequests.client;
import java.time.Duration;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.limitrequests.client.utils.Client;
import com.baeldung.limitrequests.client.utils.RandomConsumer;
import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import io.github.resilience4j.reactor.ratelimiter.operator.RateLimiterOperator;
import reactor.core.publisher.Flux;
public class Resilience4jRateLimit {
private Resilience4jRateLimit() {
}
public static Flux<Integer> fetch(WebClient client, int requests, int concurrency, int interval) {
RateLimiter limiter = RateLimiter.of("my-rate-limiter", RateLimiterConfig.custom()
.limitRefreshPeriod(Duration.ofMillis(interval))
.limitForPeriod(concurrency)
.timeoutDuration(Duration.ofMillis((long) interval * concurrency))
.build());
String clientId = Client.id(requests, Resilience4jRateLimit.class, concurrency, interval);
return Flux.range(1, requests)
.log()
.flatMap(i -> RandomConsumer.<Integer> get(client, clientId)
.transformDeferred(RateLimiterOperator.of(limiter)));
}
public static void main(String[] args) {
String baseUrl = args[0];
WebClient client = WebClient.create(baseUrl);
fetch(client, 10, 5, 2500).doOnNext(System.out::println)
.blockLast();
}
}

33
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/ZipWithInterval.java Normal file
View File

@ -0,0 +1,33 @@
package com.baeldung.limitrequests.client;
import java.time.Duration;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.limitrequests.client.utils.Client;
import com.baeldung.limitrequests.client.utils.RandomConsumer;
import reactor.core.publisher.Flux;
public class ZipWithInterval {
private ZipWithInterval() {
}
public static Flux<Integer> fetch(WebClient client, int requests, int delay) {
String clientId = Client.id(requests, ZipWithInterval.class, delay);
return Flux.range(1, requests)
.log()
.zipWith(Flux.interval(Duration.ofMillis(delay)))
.flatMap(i -> RandomConsumer.get(client, clientId));
}
public static void main(String[] args) {
String baseUrl = args[0];
WebClient client = WebClient.create(baseUrl);
fetch(client, 15, 1100).doOnNext(System.out::println)
.blockLast();
}
}

16
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/utils/Client.java Normal file
View File

@ -0,0 +1,16 @@
package com.baeldung.limitrequests.client.utils;
public interface Client {
String SEPARATOR = ":";
static String id(Object... args) {
StringBuilder builder = new StringBuilder();
for (Object object : args) {
builder.append(":")
.append(object.toString());
}
return builder.toString()
.replaceFirst(SEPARATOR, "");
}
}

19
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/client/utils/RandomConsumer.java Normal file
View File

@ -0,0 +1,19 @@
package com.baeldung.limitrequests.client.utils;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.limitrequests.server.RandomController;
import reactor.core.publisher.Mono;
public interface RandomConsumer {
static <T> Mono<T> get(WebClient client, String id) {
return client.get()
.header(RandomController.CLIENT_ID_HEADER, id)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<T>() {
});
}
}

38
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/server/Concurrency.java Normal file
View File

@ -0,0 +1,38 @@
package com.baeldung.limitrequests.server;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Concurrency {
public static final int MAX_CONCURRENT_REQUESTS = 5;
private static final Logger logger = LoggerFactory.getLogger(Concurrency.class);
private static final Map<String, AtomicInteger> CONCURRENT_REQUESTS = new HashMap<>();
private Concurrency() {
}
public static int protect(String clientId, IntSupplier supplier) throws InterruptedException {
AtomicInteger counter = CONCURRENT_REQUESTS.computeIfAbsent(clientId, k -> new AtomicInteger(0));
try {
int n = counter.incrementAndGet();
if (n > MAX_CONCURRENT_REQUESTS) {
String message = String.format("%s - %d max concurrent requests reached [%d]. try again later", clientId, MAX_CONCURRENT_REQUESTS, n);
throw new UnsupportedOperationException(message);
}
logger.info("{} - {}", clientId, n);
TimeUnit.SECONDS.sleep(2);
return supplier.getAsInt();
} finally {
counter.decrementAndGet();
}
}
}

21
spring-reactive-modules/spring-5-reactive-client-2/src/main/java/com/baeldung/limitrequests/server/RandomController.java Normal file
View File

@ -0,0 +1,21 @@
package com.baeldung.limitrequests.server;
import java.util.Random;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/random")
public class RandomController {
public static final String CLIENT_ID_HEADER = "client-id";
private static final Random RANDOM = new Random();
@GetMapping
Integer getRandom(@RequestHeader(CLIENT_ID_HEADER) String clientId) throws InterruptedException {
return Concurrency.protect(clientId, () -> RANDOM.nextInt(50));
}
}

5
spring-reactive-modules/spring-5-reactive-client-2/src/main/resources/application.properties Normal file
View File

@ -0,0 +1,5 @@
logging.level.root=INFO
server.port=8081
logging.level.reactor.netty.http.client.HttpClient=DEBUG

16
spring-reactive-modules/spring-5-reactive-client-2/src/main/resources/logback.xml Normal file
View File

@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
<layout class="ch.qos.logback.classic.PatternLayout">
# Pattern of log message for console appender
<Pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
</layout>
</appender>
<logger name="org.springframework" level="INFO" />
<root level="INFO">
<appender-ref ref="stdout" />
</root>
</configuration>

113
spring-reactive-modules/spring-5-reactive-client-2/src/test/java/com/baeldung/limitrequests/RandomControllerLiveTest.java Normal file
View File

@ -0,0 +1,113 @@
package com.baeldung.limitrequests;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertThrows;
import org.junit.jupiter.api.Test;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException.InternalServerError;
import com.baeldung.limitrequests.client.DelayElements;
import com.baeldung.limitrequests.client.GuavaRateLimit;
import com.baeldung.limitrequests.client.LimitConcurrency;
import com.baeldung.limitrequests.client.Resilience4jRateLimit;
import com.baeldung.limitrequests.client.ZipWithInterval;
import com.baeldung.limitrequests.server.Concurrency;
class RandomControllerLiveTest {
private static final int MAX_CONCURRENT_REQUESTS = Concurrency.MAX_CONCURRENT_REQUESTS;
private static final int TOTAL_REQUESTS = 10;
private WebClient client = WebClient.create("http://localhost:8081/random");
@Test
void givenEnoughDelay_whenZipWithInterval_thenNoExceptionThrown() {
int delay = MAX_CONCURRENT_REQUESTS * 100;
assertDoesNotThrow(() -> {
ZipWithInterval.fetch(client, TOTAL_REQUESTS, delay)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenSmallDelay_whenZipWithInterval_thenExceptionThrown() {
int delay = 100;
assertThrows(InternalServerError.class, () -> {
ZipWithInterval.fetch(client, TOTAL_REQUESTS, delay)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenEnoughDelay_whenDelayElements_thenNoExceptionThrown() {
int delay = MAX_CONCURRENT_REQUESTS * 100;
assertDoesNotThrow(() -> {
DelayElements.fetch(client, TOTAL_REQUESTS, delay)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenSmallDelay_whenDelayElements_thenExceptionThrown() {
int delay = 100;
assertThrows(InternalServerError.class, () -> {
DelayElements.fetch(client, TOTAL_REQUESTS, delay)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenLimitInsideServerRange_whenLimitedConcurrency_thenNoExceptionThrown() {
int limit = MAX_CONCURRENT_REQUESTS;
assertDoesNotThrow(() -> {
LimitConcurrency.fetch(client, TOTAL_REQUESTS, limit)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenLimitOutsideServerRange_whenLimitedConcurrency_thenExceptionThrown() {
int limit = MAX_CONCURRENT_REQUESTS + TOTAL_REQUESTS;
assertThrows(InternalServerError.class, () -> {
LimitConcurrency.fetch(client, TOTAL_REQUESTS, limit)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenLongInterval_whenRateLimited_thenNoExceptionThrown() {
int interval = MAX_CONCURRENT_REQUESTS * 500;
int limit = MAX_CONCURRENT_REQUESTS;
assertDoesNotThrow(() -> {
Resilience4jRateLimit.fetch(client, TOTAL_REQUESTS, limit, interval)
.doOnNext(System.out::println)
.blockLast();
});
}
@Test
void givenShortLimit_whenUsingGuavaRateLimiter_thenNoExceptionThrown() {
double limit = MAX_CONCURRENT_REQUESTS / 2;
assertDoesNotThrow(() -> {
GuavaRateLimit.fetch(client, TOTAL_REQUESTS, limit)
.doOnNext(System.out::println)
.blockLast();
});
}
}

20
spring-reactive-modules/spring-5-reactive-client-2/src/test/resources/logback-test.xml Normal file
View File

@ -0,0 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
<layout class="ch.qos.logback.classic.PatternLayout">
# Pattern of log message for console appender
<Pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
</layout>
</appender>
<logger name="org.springframework" level="INFO" />
<logger name="com.baeldung.reactive.logging.jetty" level="DEBUG" />
<logger name="reactor.netty.http.client.HttpClient" level="DEBUG" />
<logger name="com.baeldung.reactive.logging" level="DEBUG" />
<root level="INFO">
<appender-ref ref="stdout" />
</root>
</configuration>

1
spring-reactive-modules/spring-5-reactive-client/README.md
View File

@ -13,3 +13,4 @@ The "REST With Spring" Classes: http://bit.ly/restwithspring
- [Get List of JSON Objects with WebClient](https://www.baeldung.com/spring-webclient-json-list)
- [Upload a File with WebClient](https://www.baeldung.com/spring-webclient-upload-file)
- [How to Get Response Body When Testing the Status Code in WebFlux WebClient](https://www.baeldung.com/spring-webclient-get-response-body)
- More articles: [[next -->]](../spring-5-reactive-client-2)