Die Drosselungsmethode ruft M Anforderungen in N Sekunden auf

137

Ich benötige eine Komponente / Klasse, die die Ausführung einer Methode auf maximal M Aufrufe in N Sekunden drosselt (oder ms oder nanos spielt keine Rolle).

Mit anderen Worten, ich muss sicherstellen, dass meine Methode nicht mehr als M Mal in einem Schiebefenster von N Sekunden ausgeführt wird.

Wenn Sie die vorhandene Klasse nicht kennen, können Sie Ihre Lösungen / Ideen veröffentlichen, wie Sie dies implementieren würden.

vtrubnikov
quelle
3
Es gibt einige gute Antworten auf dieses Problem unter stackoverflow.com/questions/667508/…
skaffman
Ich muss sicherstellen, dass meine Methode nicht mehr als M Mal in einem Schiebefenster von N Sekunden ausgeführt wird. Ich habe kürzlich einen Blog-Beitrag darüber geschrieben, wie dies in .NET gemacht wird. Möglicherweise können Sie in Java etwas Ähnliches erstellen. Bessere Ratenbegrenzung in .NET
Jack Leitch
Die ursprüngliche Frage klingt sehr nach dem Problem, das in diesem Blog-Beitrag gelöst wurde: [Java Multi-Channel Asynchronous Throttler] ( cordinc.com/blog/2010/04/java-multichannel-asynchronous.html ). Für eine Rate von M Anrufen in N Sekunden garantiert der in diesem Blog beschriebene Throttler, dass ein Intervall der Länge N auf der Timeline nicht mehr als M Anrufe enthält.
Hbf

Antworten:

81

Ich würde einen Ringpuffer mit Zeitstempeln mit einer festen Größe von M verwenden. Bei jedem Aufruf der Methode überprüfen Sie den ältesten Eintrag. Wenn er in der Vergangenheit weniger als N Sekunden beträgt, führen Sie einen weiteren Eintrag aus und fügen ihn hinzu, andernfalls schlafen Sie für den Zeitunterschied.

Michael Borgwardt
quelle
4
Schön. Nur was ich brauche. Schnelle Versuche zeigen ~ 10 Zeilen, um diesen und minimalen Speicherbedarf zu implementieren. Sie müssen nur über die Thread-Sicherheit und das Einreihen eingehender Anforderungen nachdenken.
Vtrubnikov
5
Deshalb verwenden Sie die DelayQueue von java.util.concurrent. Es verhindert das Problem, dass mehrere Threads auf denselben Eintrag wirken.
Erickson
5
Für einen Fall mit mehreren Threads ist der Token-Bucket-Ansatz möglicherweise die bessere Wahl, denke ich.
Michael Borgwardt
1
Wissen Sie, wie dieser Algorithmus aufgerufen wird, wenn er überhaupt einen Namen hat?
Vlado Pandžić
80

Was für mich sofort funktioniert hat, war Google Guava RateLimiter .

// Allow one request per second
private RateLimiter throttle = RateLimiter.create(1.0);

private void someMethod() {
    throttle.acquire();
    // Do something
}
Schnatterer
quelle
19
Ich würde diese Lösung nicht empfehlen, da der Guava RateLimiter den Thread blockiert und dadurch den Thread-Pool leicht erschöpft.
Kaviddiss
18
@ Kaviddiss Wenn Sie nicht blockieren möchten, verwenden SietryAquire()
slf
7
Das Problem bei der aktuellen Implementierung von RateLimiter (zumindest für mich) ist, dass keine Zeiträume von mehr als 1 Sekunde und daher Raten von beispielsweise 1 pro Minute zulässig sind.
John B
4
@ John B Soweit ich weiß, können Sie mit RateLimiter 1 Anfrage pro Minute erreichen, indem Sie RateLimiter.create (60.0) + rateLimiter.acquire (60)
DivideByZero
2
@radiantRazor Ratelimiter.create (1.0 / 60) und purchase () erzielt 1 Anruf pro Minute.
Bizentass
30

Konkret sollten Sie dies mit a umsetzen können DelayQueue. Initialisieren Sie die Warteschlange mit M DelayedInstanzen, deren Verzögerung anfänglich auf Null gesetzt ist. Wenn Anforderungen an die Methode eingehen, takeein Token, das bewirkt, dass die Methode blockiert wird, bis die Drosselungsanforderung erfüllt ist. Wenn ein Token genommen wurde, wird addein neues Token mit einer Verzögerung von in die Warteschlange gestellt N.

erickson
quelle
1
Ja, das würde den Trick machen. Aber ich mag DelayQueue nicht besonders, weil es (über PriortyQueue) einen ausgeglichenen binären Hash verwendet (was viele Vergleiche offerund mögliches Array-Wachstum bedeutet), und es ist alles ziemlich schwer für mich. Ich denke für andere könnte dies vollkommen in Ordnung sein.
Vtrubnikov
5
Da in dieser Anwendung das neue Element, das dem Heap hinzugefügt wird, fast immer das maximale Element im Heap ist (dh die längste Verzögerung aufweist), ist normalerweise ein Vergleich pro Hinzufügen erforderlich. Außerdem wächst das Array niemals, wenn der Algorithmus korrekt implementiert ist, da ein Element erst nach der Aufnahme eines Elements hinzugefügt wird.
Erickson
3
Ich fand dies auch in Fällen hilfreich, in denen Sie nicht möchten, dass Anforderungen in großen Bursts auftreten, indem Sie die Größe M und die Verzögerung N in der Größenordnung von wenigen Millis relativ klein halten. z.B. M = 5, N = 20 ms würde einen Durchsatz von 250 / s Kepping Burst liefern, der in Größe 5 auftritt.
FUD
Wird diese Skala für eine Million U / min und wenn gleichzeitige Anforderungen zulässig sind? Ich müsste eine Million verzögerte Elemente hinzufügen. Auch Eckfälle haben eine hohe Latenz - Fälle, in denen mehrere Threads poll () aufrufen und jedes Mal gesperrt werden.
Aditya Joshee
@AdityaJoshee Ich habe es nicht bewertet, aber wenn ich etwas Zeit habe, werde ich versuchen, ein Gefühl für den Overhead zu bekommen. Eine Sache zu beachten ist jedoch, dass Sie nicht 1 Million Token benötigen, die in 1 Sekunde ablaufen. Sie könnten 100 Token haben, die in 10 Millisekunden ablaufen, 10 Token, die in Millisekunden ablaufen usw. Dies zwingt die Momentanrate tatsächlich dazu, näher an der Durchschnittsrate zu sein, wodurch Spitzen geglättet werden, was beim Client Backups verursachen kann, aber das ist eine natürliche Konsequenz der Ratenbegrenzung. 1 Million U / min klingt allerdings kaum nach Drosselung. Wenn Sie Ihren Anwendungsfall erläutern können, habe ich möglicherweise bessere Ideen.
Erickson
21

Informieren Sie sich über den Token-Bucket- Algorithmus. Grundsätzlich haben Sie einen Eimer mit Token darin. Jedes Mal, wenn Sie die Methode ausführen, nehmen Sie ein Token. Wenn keine Token mehr vorhanden sind, blocken Sie, bis Sie einen erhalten. In der Zwischenzeit gibt es einen externen Akteur, der die Token in einem festgelegten Intervall auffüllt.

Mir ist keine Bibliothek dafür bekannt (oder ähnliches). Sie können diese Logik in Ihren Code schreiben oder AspectJ verwenden, um das Verhalten hinzuzufügen.

Kevin
quelle
3
Danke für den Vorschlag, interessante Algo. Aber es ist nicht genau das, was ich brauche. Zum Beispiel muss ich die Ausführung auf 5 Aufrufe pro Sekunde beschränken. Wenn ich den Token-Bucket verwende und 10 Anfragen gleichzeitig eingehen, nehmen die ersten 5 Anrufe alle verfügbaren Token entgegen und werden vorübergehend ausgeführt, während die verbleibenden 5 Anrufe in einem festen Intervall von 1/5 s ausgeführt werden. In einer solchen Situation müssen die verbleibenden 5 Anrufe erst nach 1 Sekunde in einem einzigen Burst ausgeführt werden.
Vtrubnikov
5
Was ist, wenn Sie jede Sekunde 5 Token (oder 5 - (5 verbleibende)) anstelle von 1 alle 1/5 Sekunde zum Eimer hinzugefügt haben?
Kevin
@ Kevin nein das würde mir immer noch keinen "Schiebefenster" -Effekt geben
vtrubnikov
2
@valery ja das würde es. (Denken Sie daran, die Token bei M zu kappen)
Nr.
keine Notwendigkeit für einen "externen Schauspieler". Alles kann einzeln ausgeführt werden, wenn Sie Metadaten zu Anforderungszeiten aufbewahren.
Marsellus Wallace
8

Wenn Sie einen Java-basierten Schieberegler für die Schiebefensterrate benötigen, der auf einem verteilten System funktioniert, sollten Sie sich das Projekt https://github.com/mokies/ratelimitj ansehen .

Eine von Redis unterstützte Konfiguration zum Begrenzen von IP-Anforderungen auf 50 pro Minute würde folgendermaßen 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");

Weitere Informationen zur Redis-Konfiguration finden Sie unter https://github.com/mokies/ratelimitj/tree/master/ratelimitj-redis .

user2326162
quelle
5

Dies hängt von der Anwendung ab.

Stellen Sie sich den Fall , in dem mehrere Threads ein Token wollen einige tun global geschwindigkeits begrenzte Aktion mit keiner Burst erlaubt (dh Sie 10 Aktionen pro 10 Sekunden begrenzen wollen , aber Sie wollen nicht mehr als 10 Aktionen in der ersten Sekunde passieren und dann bleiben 9 Sekunden gestoppt).

Die DelayedQueue hat einen Nachteil: Die Reihenfolge, in der Threads Token anfordern, entspricht möglicherweise nicht der Reihenfolge, in der sie ihre Anforderung erfüllen. Wenn mehrere Threads blockiert sind und auf ein Token warten, ist nicht klar, welcher das nächste verfügbare Token nimmt. Aus meiner Sicht könnten sogar Threads für immer warten.

Eine Lösung besteht darin, ein Mindestintervall zwischen zwei aufeinander folgenden Aktionen einzuhalten und die Aktionen in derselben Reihenfolge auszuführen, in der 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);
        }
    }
}
Duarte Meneses
quelle
was minTimebedeutet hier Was tut es? Kannst du das erklären?
Flash
minTimeist die Mindestzeit, die nach dem Verbrauch eines Tokens vergehen muss, bevor der nächste Token verbraucht werden kann.
Duarte Meneses
3

Obwohl es nicht das ist, was Sie gefragt haben ThreadPoolExecutor, das darauf ausgelegt ist, M gleichzeitige Anforderungen anstelle von M Anforderungen in N Sekunden zu begrenzen, könnte es auch nützlich sein.

Eugene Yokota
quelle
2

Ich habe einen einfachen Drosselungsalgorithmus implementiert. Versuchen Sie diesen Link, http://krishnaprasadas.blogspot.in/2012/05/throttling-algorithm.html

Ein kurzer Überblick über den Algorithmus,

Dieser Algorithmus nutzt die Fähigkeit der Java Delayed Queue . Erstellen Sie ein verzögertes Objekt mit der erwarteten Verzögerung (hier 1000 / M für Millisekunden TimeUnit ). Stellen Sie dasselbe Objekt in die verzögerte Warteschlange, die intern das bewegliche Fenster für uns bereitstellt. Dann vor jedem Methodenaufruf nehmen das Objekt die Warteschlange bilden, nehmen ist ein blockierender Aufruf , die erst nach der angegebenen Verzögerung zurück, und nach dem Methodenaufruf nicht vergessen , mit aktualisierten Zeit das Objekt in die Warteschlange einzureihen (hier aktuelle Millisekunden) .

Hier können wir auch mehrere verzögerte Objekte mit unterschiedlicher Verzögerung haben. Dieser Ansatz bietet auch einen hohen Durchsatz.

Krishas
quelle
6
Sie sollten eine Zusammenfassung Ihres Algorithmus veröffentlichen. Wenn Ihr Link verschwindet, wird Ihre Antwort unbrauchbar.
jwr
Danke, ich habe den Brief hinzugefügt.
Krishas
1

Meine Implementierung unten kann eine beliebige Genauigkeit der Anforderungszeit verarbeiten. Sie hat eine O (1) -Zeitkomplexität für jede Anforderung, benötigt keinen zusätzlichen Puffer, z. B. O (1) -Komplexität, und erfordert keinen Hintergrundthread, um stattdessen Token freizugeben Token werden gemäß der seit der letzten Anforderung verstrichenen Zeit freigegeben.

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;
        }
    }
}
tonywl
quelle
0

Versuchen Sie diesen einfachen Ansatz zu verwenden:

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();
}

}}

SergeZ
quelle
0

Apache Camel unterstützt außerdem den folgenden Throttler- Mechanismus:

from("seda:a").throttle(100).asyncDelayed().to("seda:b");
gtonisch
quelle
0

Dies ist ein Update des obigen LeakyBucket-Codes. Dies funktioniert für mehr als 1000 Anfragen pro Sekunde.

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);
  }
}

und das Unittest für oben:

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));
  }
}
Peterreilly
quelle
Was minTimeNanobedeutet hier? können Sie erklären?
Flash
0

Hier ist eine etwas erweiterte Version des einfachen Ratenbegrenzers

/**
 * 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);
    }
}

Und Unit-Tests

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 ) );
    }

}
Leonid Astakhov
quelle
Der Code ist ziemlich einfach. Sie erstellen den Limiter einfach mit maxRate oder mit Perioden und Rate. Und dann rufen Sie einfach verbrauchen jede Anfrage. Immer wenn die Rate nicht überschritten wird, kehrt der Limiter sofort zurück oder wartet einige Zeit, bevor er zur niedrigeren aktuellen Anforderungsrate zurückkehrt. Es hat auch eine aktuelle Ratenmethode, die einen gleitenden Durchschnitt der aktuellen Rate zurückgibt.
Leonid Astakhov
0

Meine Lösung: Eine einfache util-Methode, die Sie ändern können, um eine Wrapper-Klasse zu erstellen.

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;
}

Nehmen Sie von JAVA Thread Debounce und Throttle

benbai123
quelle