Мне нужен компонент/класс, который дросселирует выполнение некоторого метода до максимальных M вызовов в N секунд (или ms или nanos, не имеет значения).
Другими словами, мне нужно убедиться, что мой метод выполняется не более M раз в скользящем окне в течение N секунд.
Если вы не знаете существующего класса, не стесняйтесь публиковать свои решения/идеи о том, как вы это реализуете.
Я бы использовал кольцевой буфер временных меток с фиксированным размером M. Каждый раз, когда вызывается метод, вы проверяете самую старую запись, и если он меньше, чем N секунд в прошлом, вы выполняете и добавляете другую запись, иначе вы спите для разницы во времени.
Для меня все было в курсе Google Guava RateLimiter.
// Allow one request per second
private RateLimiter throttle = RateLimiter.create(1.0);
private void someMethod() {
throttle.acquire();
// Do something
}
В конкретных терминах вы должны иметь возможность реализовать это с помощью DelayQueue. Инициализируйте очередь с помощью экземпляров M Delayed с первоначальной установкой задержки на ноль. В качестве запросов к методу введите take токен, который заставляет метод блокироваться до тех пор, пока не будет выполнено требование дросселирования. Когда маркер был взят, add новый токен в очередь с задержкой N.
Прочитайте алгоритм Token bucket. В принципе, у вас есть ведро с токенами. Каждый раз, когда вы выполняете метод, вы берете токен. Если больше нет токенов, вы блокируете их, пока не получите их. Между тем, есть какой-то внешний актер, который пополняет токены с фиксированным интервалом.
Я не знаю о библиотеке, чтобы сделать это (или что-то подобное). Вы можете написать эту логику в свой код или использовать AspectJ для добавления поведения.
Если вам нужен ограничитель скорости скользящего окна на основе Java, который будет работать в распределенной системе, возможно, вы захотите взглянуть на проект https://github.com/mokies/ratelimitj.
Конфигурация с поддержкой Redis, ограничивающая количество запросов по IP до 50 в минуту, будет выглядеть следующим образом:
import com.lambdaworks.redis.RedisClient;
import es.moki.ratelimitj.core.LimitRule;
RedisClient client = RedisClient.create("redis://localhost");
Set<LimitRule> rules = Collections.singleton(LimitRule.of(1, TimeUnit.MINUTES, 50)); // 50 request per minute, per key
RedisRateLimit requestRateLimiter = new RedisRateLimit(client, rules);
boolean overLimit = requestRateLimiter.overLimit("ip:127.0.0.2");
Смотрите https://github.com/mokies/ratelimitj/tree/master/ratelimitj-redis дополнительную информацию о конфигурации Redis.
Это зависит от приложения.
Представьте себе случай, когда несколько потоков хотят, чтобы токен выполнял какое-то действие с ограничением по глобальному ограничению без разрешенных пакетов (т.е. вы хотите чтобы ограничить 10 действий за 10 секунд, но вы не хотите, чтобы в первой секунде произошло 10 действий, а затем осталось 9 секунд).
Недостаток DelayedQueue: порядок, при котором токены запросов запросов не могут быть порядком, по которому они выполняют свой запрос. Если несколько потоков заблокированы в ожидании токена, неясно, какой из них будет использовать следующий доступный токен. С моей точки зрения, вы могли бы даже называть потоки.
Одним из решений является наличие минимального интервала времени между двумя последовательными действиями и выполнение действий в том же порядке, в каком они были запрошены.
Вот реализация:
public class LeakyBucket {
protected float maxRate;
protected long minTime;
//holds time of last action (past or future!)
protected long lastSchedAction = System.currentTimeMillis();
public LeakyBucket(float maxRate) throws Exception {
if(maxRate <= 0.0f) {
throw new Exception("Invalid rate");
}
this.maxRate = maxRate;
this.minTime = (long)(1000.0f / maxRate);
}
public void consume() throws InterruptedException {
long curTime = System.currentTimeMillis();
long timeLeft;
//calculate when can we do the action
synchronized(this) {
timeLeft = lastSchedAction + minTime - curTime;
if(timeLeft > 0) {
lastSchedAction += minTime;
}
else {
lastSchedAction = curTime;
}
}
//If needed, wait for our time
if(timeLeft <= 0) {
return;
}
else {
Thread.sleep(timeLeft);
}
}
}
Хотя это не то, что вы просили, ThreadPoolExecutor, который предназначен для ограничения одновременных запросов M вместо M запросов в N секунд, также может быть полезным.
Мне нужно убедиться, что мой метод выполняется не более M раз в скользящем окне N секунд.
Недавно я написал сообщение в блоге о том, как сделать это в.NET. Возможно, вы сможете создать нечто подобное в Java.
Оригинальный вопрос звучит так же, как проблема, разрешенная в этом сообщении в блоге: Java Multi-Channel Asynchronous Throttler.
Для скорости M вызовов в течение N секунд дроссельная заслонка, обсуждаемая в этом блоге, гарантирует, что любой интервал длины N на временной шкале не будет содержать больше, чем M вызовов.
Я реализовал простой алгоритм дросселирования. Попробуйте эту ссылку, http://krishnaprasadas.blogspot.in/2012/05/throttling-algorithm.html
Краткая информация об алгоритме
В этом алгоритме используется Java Delayed Queue. Создайте объект delayed с ожидаемой задержкой (здесь 1000/M для миллисекунд TimeUnit). Поместите тот же объект в задержанную очередь, которая будет предоставляться интернером для нас. Затем перед каждым вызовом метода take объект формирует очередь, возьмем блокирующий вызов, который будет возвращаться только после указанной задержки, а после вызов метода не забудьте поместить объект в очередь с обновленным временем (здесь текущие миллисекунды).
Здесь мы также можем иметь несколько задержанных объектов с разной задержкой. Такой подход также обеспечит высокую пропускную способность.
Моя реализация ниже может обрабатывать произвольную точность времени запроса, она имеет O (1) временную сложность для каждого запроса, не требует никакого дополнительного буфера, например O (1) сложность пространства, кроме того, для освобождения токена не требуется фоновый поток, вместо этого токены освобождаются в соответствии с временем, прошедшим с момента последнего запроса.
class RateLimiter {
int limit;
double available;
long interval;
long lastTimeStamp;
RateLimiter(int limit, long interval) {
this.limit = limit;
this.interval = interval;
available = 0;
lastTimeStamp = System.currentTimeMillis();
}
synchronized boolean canAdd() {
long now = System.currentTimeMillis();
// more token are released since last request
available += (now-lastTimeStamp)*1.0/interval*limit;
if (available>limit)
available = limit;
if (available<1)
return false;
else {
available--;
lastTimeStamp = now;
return true;
}
}
}
Попробуйте использовать этот простой подход:
public class SimpleThrottler {
private static final int T = 1; // min
private static final int N = 345;
private Lock lock = new ReentrantLock();
private Condition newFrame = lock.newCondition();
private volatile boolean currentFrame = true;
public SimpleThrottler() {
handleForGate();
}
/**
* Payload
*/
private void job() {
try {
Thread.sleep(Math.abs(ThreadLocalRandom.current().nextLong(12, 98)));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.err.print(" J. ");
}
public void doJob() throws InterruptedException {
lock.lock();
try {
while (true) {
int count = 0;
while (count < N && currentFrame) {
job();
count++;
}
newFrame.await();
currentFrame = true;
}
} finally {
lock.unlock();
}
}
public void handleForGate() {
Thread handler = new Thread(() -> {
while (true) {
try {
Thread.sleep(1 * 900);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
currentFrame = false;
lock.lock();
try {
newFrame.signal();
} finally {
lock.unlock();
}
}
}
});
handler.start();
}
}
Apache Camel также поддерживает Throttler механизма следующим образом:
from("seda:a").throttle(100).asyncDelayed().to("seda:b");
Вы можете использовать redis для этого, когда блокировка необходима в распределенной системе. Второй алгоритм в https://redis.io/commands/incr
Это обновление для кода LeakyBucket выше. Это работает для более 1000 запросов в секунду.
import lombok.SneakyThrows;
import java.util.concurrent.TimeUnit;
class LeakyBucket {
private long minTimeNano; // sec / billion
private long sched = System.nanoTime();
/**
* Create a rate limiter using the leakybucket alg.
* @param perSec the number of requests per second
*/
public LeakyBucket(double perSec) {
if (perSec <= 0.0) {
throw new RuntimeException("Invalid rate " + perSec);
}
this.minTimeNano = (long) (1_000_000_000.0 / perSec);
}
@SneakyThrows public void consume() {
long curr = System.nanoTime();
long timeLeft;
synchronized (this) {
timeLeft = sched - curr + minTimeNano;
sched += minTimeNano;
}
if (timeLeft <= minTimeNano) {
return;
}
TimeUnit.NANOSECONDS.sleep(timeLeft);
}
}
и unittest для выше:
import com.google.common.base.Stopwatch;
import org.junit.Ignore;
import org.junit.Test;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
public class LeakyBucketTest {
@Test @Ignore public void t() {
double numberPerSec = 10000;
LeakyBucket b = new LeakyBucket(numberPerSec);
Stopwatch w = Stopwatch.createStarted();
IntStream.range(0, (int) (numberPerSec * 5)).parallel().forEach(
x -> b.consume());
System.out.printf("%,d ms%n", w.elapsed(TimeUnit.MILLISECONDS));
}
}
Вот немного продвинутая версия простого ограничителя скорости
/**
* Simple request limiter based on Thread.sleep method.
* Create limiter instance via {@link #create(float)} and call {@link #consume()} before making any request.
* If the limit is exceeded cosume method locks and waits for current call rate to fall down below the limit
*/
public class RequestRateLimiter {
private long minTime;
private long lastSchedAction;
private double avgSpent = 0;
ArrayList<RatePeriod> periods;
@AllArgsConstructor
public static class RatePeriod{
@Getter
private LocalTime start;
@Getter
private LocalTime end;
@Getter
private float maxRate;
}
/**
* Create request limiter with maxRate - maximum number of requests per second
* @param maxRate - maximum number of requests per second
* @return
*/
public static RequestRateLimiter create(float maxRate){
return new RequestRateLimiter(Arrays.asList( new RatePeriod(LocalTime.of(0,0,0),
LocalTime.of(23,59,59), maxRate)));
}
/**
* Create request limiter with ratePeriods calendar - maximum number of requests per second in every period
* @param ratePeriods - rate calendar
* @return
*/
public static RequestRateLimiter create(List<RatePeriod> ratePeriods){
return new RequestRateLimiter(ratePeriods);
}
private void checkArgs(List<RatePeriod> ratePeriods){
for (RatePeriod rp: ratePeriods ){
if ( null == rp || rp.maxRate <= 0.0f || null == rp.start || null == rp.end )
throw new IllegalArgumentException("list contains null or rate is less then zero or period is zero length");
}
}
private float getCurrentRate(){
LocalTime now = LocalTime.now();
for (RatePeriod rp: periods){
if ( now.isAfter( rp.start ) && now.isBefore( rp.end ) )
return rp.maxRate;
}
return Float.MAX_VALUE;
}
private RequestRateLimiter(List<RatePeriod> ratePeriods){
checkArgs(ratePeriods);
periods = new ArrayList<>(ratePeriods.size());
periods.addAll(ratePeriods);
this.minTime = (long)(1000.0f / getCurrentRate());
this.lastSchedAction = System.currentTimeMillis() - minTime;
}
/**
* Call this method before making actual request.
* Method call locks until current rate falls down below the limit
* @throws InterruptedException
*/
public void consume() throws InterruptedException {
long timeLeft;
synchronized(this) {
long curTime = System.currentTimeMillis();
minTime = (long)(1000.0f / getCurrentRate());
timeLeft = lastSchedAction + minTime - curTime;
long timeSpent = curTime - lastSchedAction + timeLeft;
avgSpent = (avgSpent + timeSpent) / 2;
if(timeLeft <= 0) {
lastSchedAction = curTime;
return;
}
lastSchedAction = curTime + timeLeft;
}
Thread.sleep(timeLeft);
}
public synchronized float getCuRate(){
return (float) ( 1000d / avgSpent);
}
}
И юнит-тесты
import org.junit.Assert;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class RequestRateLimiterTest {
@Test(expected = IllegalArgumentException.class)
public void checkSingleThreadZeroRate(){
// Zero rate
RequestRateLimiter limiter = RequestRateLimiter.create(0);
try {
limiter.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Test
public void checkSingleThreadUnlimitedRate(){
// Unlimited
RequestRateLimiter limiter = RequestRateLimiter.create(Float.MAX_VALUE);
long started = System.currentTimeMillis();
for ( int i = 0; i < 1000; i++ ){
try {
limiter.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long ended = System.currentTimeMillis();
System.out.println( "Current rate:" + limiter.getCurRate() );
Assert.assertTrue( ((ended - started) < 1000));
}
@Test
public void rcheckSingleThreadRate(){
// 3 request per minute
RequestRateLimiter limiter = RequestRateLimiter.create(3f/60f);
long started = System.currentTimeMillis();
for ( int i = 0; i < 3; i++ ){
try {
limiter.consume();
Thread.sleep(20000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long ended = System.currentTimeMillis();
System.out.println( "Current rate:" + limiter.getCurRate() );
Assert.assertTrue( ((ended - started) >= 60000 ) & ((ended - started) < 61000));
}
@Test
public void checkSingleThreadRateLimit(){
// 100 request per second
RequestRateLimiter limiter = RequestRateLimiter.create(100);
long started = System.currentTimeMillis();
for ( int i = 0; i < 1000; i++ ){
try {
limiter.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long ended = System.currentTimeMillis();
System.out.println( "Current rate:" + limiter.getCurRate() );
Assert.assertTrue( (ended - started) >= ( 10000 - 100 ));
}
@Test
public void checkMultiThreadedRateLimit(){
// 100 request per second
RequestRateLimiter limiter = RequestRateLimiter.create(100);
long started = System.currentTimeMillis();
List<Future<?>> tasks = new ArrayList<>(10);
ExecutorService exec = Executors.newFixedThreadPool(10);
for ( int i = 0; i < 10; i++ ) {
tasks.add( exec.submit(() -> {
for (int i1 = 0; i1 < 100; i1++) {
try {
limiter.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}) );
}
tasks.stream().forEach( future -> {
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
long ended = System.currentTimeMillis();
System.out.println( "Current rate:" + limiter.getCurRate() );
Assert.assertTrue( (ended - started) >= ( 10000 - 100 ) );
}
@Test
public void checkMultiThreaded32RateLimit(){
// 0,2 request per second
RequestRateLimiter limiter = RequestRateLimiter.create(0.2f);
long started = System.currentTimeMillis();
List<Future<?>> tasks = new ArrayList<>(8);
ExecutorService exec = Executors.newFixedThreadPool(8);
for ( int i = 0; i < 8; i++ ) {
tasks.add( exec.submit(() -> {
for (int i1 = 0; i1 < 2; i1++) {
try {
limiter.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}) );
}
tasks.stream().forEach( future -> {
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
long ended = System.currentTimeMillis();
System.out.println( "Current rate:" + limiter.getCurRate() );
Assert.assertTrue( (ended - started) >= ( 10000 - 100 ) );
}
@Test
public void checkMultiThreadedRateLimitDynamicRate(){
// 100 request per second
RequestRateLimiter limiter = RequestRateLimiter.create(100);
long started = System.currentTimeMillis();
List<Future<?>> tasks = new ArrayList<>(10);
ExecutorService exec = Executors.newFixedThreadPool(10);
for ( int i = 0; i < 10; i++ ) {
tasks.add( exec.submit(() -> {
Random r = new Random();
for (int i1 = 0; i1 < 100; i1++) {
try {
limiter.consume();
Thread.sleep(r.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}) );
}
tasks.stream().forEach( future -> {
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
long ended = System.currentTimeMillis();
System.out.println( "Current rate:" + limiter.getCurRate() );
Assert.assertTrue( (ended - started) >= ( 10000 - 100 ) );
}
}
Мое решение: простой метод util, вы можете изменить его для создания класса-оболочки.
public static Runnable throttle (Runnable realRunner, long delay) {
Runnable throttleRunner = new Runnable() {
// whether is waiting to run
private boolean _isWaiting = false;
// target time to run realRunner
private long _timeToRun;
// specified delay time to wait
private long _delay = delay;
// Runnable that has the real task to run
private Runnable _realRunner = realRunner;
@Override
public void run() {
// current time
long now;
synchronized (this) {
// another thread is waiting, skip
if (_isWaiting) return;
now = System.currentTimeMillis();
// update time to run
// do not update it each time since
// you do not want to postpone it unlimited
_timeToRun = now+_delay;
// set waiting status
_isWaiting = true;
}
try {
Thread.sleep(_timeToRun-now);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// clear waiting status before run
_isWaiting = false;
// do the real task
_realRunner.run();
}
}};
return throttleRunner;
}
Взять из JAVA Thread Debounce и Throttle
Ознакомьтесь с классом [TimerTask 1. Или ScheduledExecutor.