Die Drosselungsmethode ruft M Anforderungen in N Sekunden auf

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.

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.

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
}

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.

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.

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 .

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

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.

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.

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

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

}}

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

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

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

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

}

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