Drosselung Methode aufruft, um M-Anfragen in N Sekunden
-
05-07-2019 - |
Frage
Ich brauche eine Komponente / Klasse, die Ausführung einiger Verfahren drosselt bis maximal M in N Sekunden-Anrufe (oder ms oder nanos, spielt keine Rolle).
Mit anderen Worten muß ich sicherstellen, dass meine Methode nicht mehr als M-mal in einem Schiebefenster von N Sekunden ausgeführt wird.
Wenn Sie nicht wissen, bestehende Klasse fühlen sich frei, Ihre Lösungen / Ideen zu schreiben, wie Sie dies umsetzen würde.
Lösung
würde ich einen Ringpuffer der Zeitstempel mit einer festen Größe von M. verwenden Jedes Mal, die Methode aufgerufen wird, den ältesten Eintrag zu überprüfen, und wenn es weniger als N Sekunden in der Vergangenheit Sie ausführen und eine weiteren Eintrag hinzufügen, andernfalls Sie für die Zeitdifferenz schlafen.
Andere Tipps
Was für mich aus dem Feld arbeitete, war Google Guava RateLimiter .
// Allow one request per second
private RateLimiter throttle = RateLimiter.create(1.0);
private void someMethod() {
throttle.acquire();
// Do something
}
Konkret sollten Sie in der Lage sein, dies zu implementieren mit einem Lesen Sie auf der Token Eimer Algorithmus. Grundsätzlich haben Sie einen Eimer mit Token in ihm. Jedes Mal, wenn Sie die Methode ausführen, nehmen Sie ein Token. Wenn es keine weiteren Token sind, blockieren Sie, bis Sie sich ein. Inzwischen gibt es einig externen Akteur, der die Token in einem festen Intervall wieder auffüllt. Ich bin mir nicht bewusst, eine Bibliothek, dies zu tun (oder etwas ähnliches). Man könnte diese Logik in den Code schreiben oder AspectJ verwendet das Verhalten hinzuzufügen. DelayQueue
. Initialisieren der Warteschlange mit M
Delayed
Instanzen mit ihrer Verzögerung anfangs auf Null gesetzt. Als Anforderungen an den Verfahren kommen in take
ein Token, welches das Verfahren veranlasst, bis die Drosselanforderung erfüllt wurde zu blockieren. Wenn ein Token genommen wurde,
Dies hängt in der Anwendung.
Stellen Sie sich den Fall, in dem mehr Threads wollen ein Zeichen etwas global Rate begrenzte Aktion mit ohne Burst zu tun erlaubt (dh Sie wollen 10 Aktionen pro 10 Sekunden zu begrenzen, aber Sie wollen nicht mehr als 10 Aktionen in der ersten Sekunde passieren und verbleiben dann 9 Sekunden gestoppt).
Die DelayedQueue hat einen Nachteil: Die Reihenfolge, bei der Anfrage Token fädelt der Auftrag nicht sein könnte, an dem sie ihre Anfrage bekommen erfüllt. Wenn mehrere Threads für ein Token-blocked wartet, ist es nicht klar, welche nimmt einer die nächsten verfügbaren Token. Man könnte sogar in meiner Sicht Threads ewig warten.
Eine Lösung ist hat ein Mindestzeitintervall zwischen zwei aufeinanderfolgenden Aktionen , und Aktionen in der gleichen Reihenfolge, wie sie angefordert wurden.
Hier ist eine Implementierung:
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);
}
}
}
Wenn Sie eine Java-basierte Schiebefenster Geschwindigkeitsbegrenzer müssen, die in einem verteilten System arbeiten werden Sie vielleicht ein href nehmen wollen einen Blick auf die <= „https://github.com/mokies/ratelimitj“ rel = "noreferrer „> https://github.com/mokies/ratelimitj Projekt.
Eine Redis unterstützt Konfiguration, bis 50 pro Minute Anfragen von IP begrenzen würde wie folgt aussehen:
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");
Siehe https://github.com/mokies/ratelimitj/tree/master / ratelimitj-redis Vordergrund weitere Details zu Redis Konfiguration.
Es ist zwar nicht das, was Sie gefragt,
ich eine einfache Drosselung algorithm.Try Link implementiert haben,
http://krishnaprasadas.blogspot.in/2012/05/throttling-algorithm.html Eine kurze über den Algorithmus, Dieser Algorithmus nutzt die Fähigkeit von Java Verzögerte Warteschlange .
Erstellen Sie eine verzögert mit dem erwarteten Objekt Verzögerung (hier 1000 / M für Millisekunden Timeunit ).
Legen Sie das gleiche Objekt in die verzögerte Warteschlange, die wird intern das bewegliche Fenster für uns zur Verfügung stellt.
Dann vor jedem Methodenaufruf
unten Meine Implementierung beliebige Anforderungszeit Präzision verarbeiten kann, hat es O (1) Zeitkomplexität für jede Anforderung, erfordert keine zusätzlichen Puffer, z.B. O (1) Speicherkomplexität, zusätzlich spielt es keinen Hintergrund-Thread benötigt Token freizugeben, anstatt Token freigegeben wird nach Zeit seit dem letzten Anforderung übergeben. Versuchen Sie diesen einfachen Ansatz zu verwenden: } Apache Camel unterstützt auch kommt mit Throttler Mechanismus wie folgt: Dies ist ein Update für den Leaky-Bucket-Algorithmus Code oben.
Dies funktioniert für eine mehr als 1000 Anfragen pro Sekunde. und der Unittest für oben: Hier ist eine kleine erweiterte Version von einfacher Begrenzung Und Unit-Tests 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();
}
from("seda:a").throttle(100).asyncDelayed().to("seda:b");
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);
}
}
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 ) );
}
}