doc: 阅读RateLimiter文档
This commit is contained in:
asahi
@@ -1,3 +1,39 @@
|
||||
- [CircuitBreaker](#circuitbreaker)
|
||||
- [Introduction](#introduction)
|
||||
- [maven](#maven)
|
||||
- [CircuitBreaker](#circuitbreaker-1)
|
||||
- [State Machine](#state-machine)
|
||||
- [Sliding Window](#sliding-window)
|
||||
- [Count-based sliding window](#count-based-sliding-window)
|
||||
- [Time-based sliding window](#time-based-sliding-window)
|
||||
- [Failure rate and slow call rate thresholds](#failure-rate-and-slow-call-rate-thresholds)
|
||||
- [Failure rate \& exception list](#failure-rate--exception-list)
|
||||
- [Slow call rate](#slow-call-rate)
|
||||
- [CircuitBreaker in `OPEN`/`HALF_OPEN` state](#circuitbreaker-in-openhalf_open-state)
|
||||
- [Special States](#special-states)
|
||||
- [thread-safe](#thread-safe)
|
||||
- [CircuitBreakerRegistry](#circuitbreakerregistry)
|
||||
- [Create and configure CircuitBreakerConfig](#create-and-configure-circuitbreakerconfig)
|
||||
- [success/failure/ignore判断流程](#successfailureignore判断流程)
|
||||
- [Decorate and execute a functional interface](#decorate-and-execute-a-functional-interface)
|
||||
- [Consume emitted RegistryEvents](#consume-emitted-registryevents)
|
||||
- [consume emitted CircuitBreakerEvents](#consume-emitted-circuitbreakerevents)
|
||||
- [Bulkhead](#bulkhead)
|
||||
- [create a BulkheadRegistry](#create-a-bulkheadregistry)
|
||||
- [Create and configure a Bulkhead](#create-and-configure-a-bulkhead)
|
||||
- [Create and configure a ThreadPoolBulkhead](#create-and-configure-a-threadpoolbulkhead)
|
||||
- [decorate and execute a functional interface](#decorate-and-execute-a-functional-interface-1)
|
||||
- [consume emitted RegistryEvents](#consume-emitted-registryevents-1)
|
||||
- [consume emitted BulkheadEvents](#consume-emitted-bulkheadevents)
|
||||
- [ThreadPoolBulkhead弊端](#threadpoolbulkhead弊端)
|
||||
- [RateLimiter](#ratelimiter)
|
||||
- [Create RateLimiterRegistry](#create-ratelimiterregistry)
|
||||
- [Create and configure a RateLimiter](#create-and-configure-a-ratelimiter)
|
||||
- [decorate and execute a functional interface](#decorate-and-execute-a-functional-interface-2)
|
||||
- [consume emitted RegistryEvents](#consume-emitted-registryevents-2)
|
||||
- [consume emitted RateLimiterEvents](#consume-emitted-ratelimiterevents)
|
||||
|
||||
|
||||
# CircuitBreaker
|
||||
resilience4j是一个轻量级的fault tolerance library，其针对函数式编程进行设计。resilence4j提供了更高阶的函数（`decorator`）来对`function interface, lambda expression, method reference`等内容进行增强。
|
||||
|
||||
@@ -564,3 +600,107 @@ bulkhead.getEventPublisher()
|
||||
.onCallRejected(event -> logger.info(...))
|
||||
.onCallFinished(event -> logger.info(...));
|
||||
```
|
||||
|
||||
### ThreadPoolBulkhead弊端
|
||||
在ThreadPoolBulkhead的实现中，会为每一个bulkhead都创建独立的线程池，故而应当避免在项目中创建大量的bulkhead，避免项目线程数量的膨胀以及线程切换带来的巨大开销。
|
||||
|
||||
## RateLimiter
|
||||
Resilience4j提供了RateLimiter，其将从`epoch`（jvm启动开始）开始的所有nanos划分为了一系列周期，每个周期的时长可以通过`RateLimiterConfig.limitRefreshPeriod`来进行配置。在每个周期的开始，RateLimiter都会将`active permissions number`设置为`RateLimiterConfig.limitForPeriod`。
|
||||
|
||||
RateLimiter的默认实现为`AtomicRateLimiter`，其通过`AtomicReference`来管理自身的状态。`AtomicRateLimiter.State`其本身是不可变的，并且包含如下fields:
|
||||
- `activeCycle`: 上次调用所使用的cycle number
|
||||
- `activePermissions`： 在上次调用之后的available permissions，该field可以为负数
|
||||
- `nanosToWait`：为了最后一次调用所需要等待的nanos数
|
||||
|
||||
除了`AtomicRateLimiter`之外，还存在`SemaphoreBasedRateLimiter`实现，其使用了semaphore和scheduler在每次`RatelImiterConfig#limitRefreshPeriod`之后对permissions进行刷新
|
||||
|
||||
### Create RateLimiterRegistry
|
||||
和CircuitBreaker module类似，RateLimiter module也提供了in-memory RateLimiterRegistry用于管理RateLimiter实例。
|
||||
|
||||
```java
|
||||
RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.ofDefaults();
|
||||
```
|
||||
|
||||
### Create and configure a RateLimiter
|
||||
可以提供自定义的`RateLimiterConfig`，可通过builder进行创建。`RateLimiterConfig`支持如下配置属性：
|
||||
- `timeoutDuration`:
|
||||
- `default value`: 5 [s]
|
||||
- `description`: 线程等待permission的默认等待时间
|
||||
- `limitRefreshPeriod`:
|
||||
- `default value`: 500 [ns]
|
||||
- `description`: limit refresh period。在每个period刷新后，rate limiter都会将其permission count重新设置为`limitForPeriod`
|
||||
- `limitForPeriod`:
|
||||
- `default value`: 50
|
||||
- `description`: 一个refresh period内的可获取permissions数量
|
||||
|
||||
通过rateLimiter限制某些接口调用速率不能超过`10 req/ms`的示例如下：
|
||||
```java
|
||||
RateLimiterConfig config = RateLimiterConfig.custom()
|
||||
.limitRefreshPeriod(Duration.ofMillis(1))
|
||||
.limitForPeriod(10)
|
||||
.timeoutDuration(Duration.ofMillis(25))
|
||||
.build();
|
||||
|
||||
// Create registry
|
||||
RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(config);
|
||||
|
||||
// Use registry
|
||||
RateLimiter rateLimiterWithDefaultConfig = rateLimiterRegistry
|
||||
.rateLimiter("name1");
|
||||
|
||||
RateLimiter rateLimiterWithCustomConfig = rateLimiterRegistry
|
||||
.rateLimiter("name2", config);
|
||||
```
|
||||
|
||||
### decorate and execute a functional interface
|
||||
RateLimiter支持对`callable, supplier, runnable, consumer, checkedRunnalbe, checkedSupplier, checkedConsumer, CompletionStage`进行decorate：
|
||||
```java
|
||||
// Decorate your call to BackendService.doSomething()
|
||||
CheckedRunnable restrictedCall = RateLimiter
|
||||
.decorateCheckedRunnable(rateLimiter, backendService::doSomething);
|
||||
|
||||
Try.run(restrictedCall)
|
||||
.andThenTry(restrictedCall)
|
||||
.onFailure((RequestNotPermitted throwable) -> LOG.info("Wait before call it again :)"));
|
||||
```
|
||||
可以使用`changeTimeoutDuration`和`changeLimitForPeriod`在运行时改变rateLimiter的参数。new timeout并不会影响`正在等待permissions`的线程。并且，new limit不会影响当前period的permissions，new limiter会从下个period开始应用：
|
||||
```java
|
||||
// Decorate your call to BackendService.doSomething()
|
||||
CheckedRunnable restrictedCall = RateLimiter
|
||||
.decorateCheckedRunnable(rateLimiter, backendService::doSomething);
|
||||
|
||||
// during second refresh cycle limiter will get 100 permissions
|
||||
rateLimiter.changeLimitForPeriod(100);
|
||||
```
|
||||
|
||||
### consume emitted RegistryEvents
|
||||
可以针对`RateLimiterRegistry`注册event consumer，对rateLimiter的创建、替换和删除事件进行监听：
|
||||
```java
|
||||
RateLimiterRegistry registry = RateLimiterRegistry.ofDefaults();
|
||||
registry.getEventPublisher()
|
||||
.onEntryAdded(entryAddedEvent -> {
|
||||
RateLimiter addedRateLimiter = entryAddedEvent.getAddedEntry();
|
||||
LOG.info("RateLimiter {} added", addedRateLimiter.getName());
|
||||
})
|
||||
.onEntryRemoved(entryRemovedEvent -> {
|
||||
RateLimiter removedRateLimiter = entryRemovedEvent.getRemovedEntry();
|
||||
LOG.info("RateLimiter {} removed", removedRateLimiter.getName());
|
||||
});
|
||||
```
|
||||
### consume emitted RateLimiterEvents
|
||||
RateLimiter会在如下场景下发送事件：
|
||||
- a successful permission acquire
|
||||
- acquire failure
|
||||
|
||||
所有的事件都包含额外信息，例如事件创建事件和rate limiter name等。如果想要消费事件，可以针对rateLimiter注册event consumer：
|
||||
```java
|
||||
rateLimiter.getEventPublisher()
|
||||
.onSuccess(event -> logger.info(...))
|
||||
.onFailure(event -> logger.info(...));
|
||||
```
|
||||
如果使用project reactor，可以使用如下方式进行注册：
|
||||
```java
|
||||
ReactorAdapter.toFlux(rateLimiter.getEventPublisher())
|
||||
.filter(event -> event.getEventType() == FAILED_ACQUIRE)
|
||||
.subscribe(event -> logger.info(...))
|
||||
```
|
||||
|
||||
Reference in New Issue
Block a user