Construire un limiteur de débit simple en Java | de Vishal Ratna | janvier 2023

Avec des utilisations du point de vue d’Android

Les utilisateurs du monde réel sont cruels, impatients et sournois. Dans les systèmes à forte concurrence, il peut y avoir des situations où votre serveur peut être bombardé de fausses requêtes, vous voudrez peut-être contrôler cela.

Certains des cas d’utilisation réels pourraient être les suivants :

1. Gestion des quotas d’API – En tant que fournisseur, vous souhaiterez peut-être limiter le taux auquel les demandes d’API sont adressées à votre serveur, en fonction des plans de paiement que l’utilisateur a pris. Cela peut être du côté du client ou du service.

2. Sécurité — Pour se protéger contre les attaques DDOS.

3. Contrôle des coûts – Ce n’est pas nécessairement pour le côté service ou même le côté client. Si un composant émet un événement à un taux très élevé, cela pourrait aider à le contrôler. Cela pourrait aider à contrôler la télémétrie envoyée du côté client.

Options que nous avons tout en limitant le débit

Selon le type de demande/d’événement auquel nous sommes confrontés, les événements suivants peuvent se produire :

1. Nous pouvons supprimer les demandes supplémentaires
2. Nous pouvons choisir de faire attendre les demandes jusqu’à ce que le système les limite au taux prédéfini.

Algorithmes communs de limitation de débit

1. Algorithme de seau à jetons
2. Algorithme de seau qui fuit

Nous n’entrerons pas dans les détails internes de ces algorithmes car cela n’entre pas dans le cadre de cet article.

Nous prendrons l’algorithme du seau à jetons comme pivot. Écrivons les exigences de haut niveau pour la mise en œuvre.

L’algorithme de seau à jetons est basé sur une analogie d’un seau à capacité fixe qui ajoute des jetons à un taux fixe. Avant d’autoriser une API à continuer, le compartiment est inspecté pour voir s’il contient au moins un jeton à ce moment-là. Si le jeton existe, un appel API est effectué. Sinon, il est abandonné/ou mis en attente.

1. Doit être en mesure d’accepter les transactions requises (TPS) par seconde ou le taux.
2. Devrait abandonner les transactions si elles dépassent notre taux défini.
3. Devrait fonctionner dans des situations concurrentes.

1. Doit être capable de lisser les rafales de demandes. Par exemple, si nous avons défini TPS comme 5, et que les cinq requêtes arrivent au même moment, il doit pouvoir les aligner à intervalles fixes, c’est-à-dire exécuter chacune d’elles à un intervalle de 200 ms. Il nécessite un circuit de temporisation interne.
2. Si nous avons un TPS de 5, et dans l’un des créneaux de 1 seconde, nous n’utilisons que trois jetons pour la seconde suivante, nous devrions être capables de fournir 5 + 2 = 7 jetons en bonus. Mais au rythme de 1/7 (142,28 ms) par jeton. Le bonus ne doit pas être reporté au créneau suivant.

Définissons d’abord le contrat de notre RateLimiter:

/**
* Rate limiter helps in limiting the rate of execution of a piece of code. The rate is defined in terms of
* TPS(Transactions per second). Rate of 5 would suggest, 5 transactions/second. Transaction could be a DB call, API call,
* or a simple function call.
* <p>
* Every {@link RateLimiter} implementation should implement either {@link RateLimiter#throttle(Code)} or, {@link RateLimiter#enter()}.
* They can also choose to implement all.
* <p>
* {@link Code} represents a piece of code that needs to be rate limited. It could be a function call, if the code to be rate limited
* spreads across multiple functions, we need to use entry() and exit() contract.
*/
public interface RateLimiter {
/**
* Rate limits the code passed inside as an argument.
*
* @param code representation of the piece of code that needs to be rate limited.
* @return true if executed, false otherwise.
*/
boolean throttle(Code code);
/**
* When the piece of code that needs to be rate limited cannot be represented as a contiguous
* code, then entry() should be used before we start executing the code. This brings the code inside the rate
* limiting boundaries.
*
* @return true if the code will execute and false if rate limited.
* <p
*/
boolean enter();
/**
* Interface to represent a contiguous piece of code that needs to be rate limited.
*/
interface Code {
/**
* Calling this function should execute the code that is delegated to this interface.
*/
void invoke();
}
}

Notre RateLimiter a deux ensembles d’API : l’un est throttle(code)et un autre est enter(). Les deux répondent à la même fonctionnalité mais de ces deux manières :

1. boolean throttle(code) — peut être utilisé pour passer un bloc de code si nous avons du code contigu avec nous.
2. boolean enter() – peut être utilisé en général avant l’API, la base de données ou tout appel que nous voulons limiter. Si le code suivant s’exécutait, il renverrait trueet false s’il est limité en débit. Vous pouvez mettre ces demandes en file d’attente ou les rejeter.

Vous ne verrez jamais le throttle(code) mise en œuvre en production, car elle est sous-optimale. S’il vous plaît laissez-moi savoir pourquoi dans les commentaires. La plupart des limiteurs de débit utilisent les API qui ressemblent à enter().

Pour construire le cœur de notre limiteur de débit, nous devons nous assurer qu’entre deux secondes, nous ne devons pas autoriser plus de N transactions. Comment allons-nous faire cela?

Considérez le moment où nous effectuons la première transaction t0. Donc,
jusqu’à ce que (t0 + 1)snous sommes autorisés à faire seulement N transactions. Et comment va-t-on s’en assurer ? Lors de la prochaine transaction, nous vérifierons si
current time ≤ (t0+1). Sinon, cela signifie que nous sommes entrés dans une seconde différente, et nous sommes autorisés à faire N transactions. Voyons une petite section de code qui démontre que :

long now = System.nanoTime();
if (now <= mNextSecondBoundary) { // If we are within the time limit of current second
if (mCounter < N) { // If token available
mLastExecutionNanos = now;
mCounter++; // Allocate token
invoke(code); // Invoke the code passed the throttle method.
}
}

Alors, comment définit-on le mNextSecondBoundary? Cela sera fait lorsque nous effectuerons la première transaction, comme indiqué, nous ajouterons une seconde au moment où la première transaction sera effectuée.

if (mLastExecutionNanos == 0L) {
mCounter++; // Allocate the very first token here.
mLastExecutionNanos = System.nanoTime();
mNextSecondBoundary = mLastExecutionNanos + NANO_PER_SEC;  // (10^9)
}

Maintenant, que devons-nous faire si nous exécutons le code et constatons que nous sommes entrés dans une seconde différente ? Nous allons améliorer le code précédent en réinitialisant notre dernière heure d’exécution, le nombre de jetons disponibles et en répétant le même processus en appelant throttle() de nouveau. Notre méthode sait déjà gérer une nouvelle seconde.

@Override
public boolean throttle(Code code) {
if (mTPS <= 0) {
// We do not want anything to pass.
return false;
}
synchronized (mLock) {
if (mLastExecutionNanos == 0L) {
mCounter++;
mLastExecutionNanos = System.nanoTime();
mNextSecondBoundary = mLastExecutionNanos + NANO_PER_SEC;
invoke(code);
return true;
} else {
long now = System.nanoTime();
if (now <= mNextSecondBoundary) {
if (mCounter < mTPS) {
mLastExecutionNanos = now;
mCounter++;
invoke(code);
return true;
} else {
return false;
}
} else {
// Reset the counter as we in a different second now.
mCounter = 0;
mLastExecutionNanos = 0L;
mNextSecondBoundary = 0L;
return throttle(code);
}
}
}
}

Dans cette implémentation, nous pouvons passer le bloc de code qui doit être limité, mais il y a un problème avec ce code. Cela fonctionnera mais il fonctionnera mal. Ce n’est pas recommandé, mais pourquoi ? S’il vous plaît laissez-moi savoir dans les commentaires.

Il est maintenant temps de créer la deuxième API en utilisant les mêmes blocs de construction et enter(). Nous utiliserons la même logique, mais nous n’allons pas exécuter le bloc de code à l’intérieur de la méthode. Il sera plutôt suivi après l’appel à enter() comme nous le faisons la gestion de l’Etat. La mise en œuvre de la méthode est la suivante :

@Override
public boolean enter() {
if (mTPS == 0L) {
return false;
}
synchronized (mBoundaryLock) {
if (mLastExecutionNanos == 0L) {
mLastExecutionNanos = System.nanoTime();
mCounter++;
mNextSecondBoundary = mLastExecutionNanos + NANO_PER_SEC;
return true;
} else {
long now = System.nanoTime();
if (now <= mNextSecondBoundary) {
if (mCounter < mTPS) {
mLastExecutionNanos = now;
mCounter++;
return true;
} else return false;
} else {
// Reset the counter as we in a different second now.
mCounter = 0;
mLastExecutionNanos = 0L;
mNextSecondBoundary = 0L;
return enter();
}
}
}
}

Maintenant, notre simple limiteur de débit est prêt à l’emploi. Vous pouvez consulter le code complet ici.

Nous allons essayer de créer un code de pilote qui crée six threads. Chaque thread essaie de compter de 0 à 100 avec un délai de 50 ms (qui peut être défini sur n’importe quel nombre). Nous mettrons notre limiteur de débit en action comme suit :

public static void main(String[] args) {
RateLimiter limiter = new SimpleTokenBucketRateLimiter(1);
Thread[] group = new Thread[6];
Runnable r = () -> {
for (int i = 0; i < 100; i++) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
if (limiter.enter()) {
System.out.println("Values:- " + Thread.currentThread().getName() + ": " + i);
}
}
};
for (int i = 0; i < 6; i++) {
group[i] = new Thread(r);
group[i].start();
}
}

Nos API ne prennent pas en charge le lissage des transactions, mais à la place, elles les font attendre que le prochain jeton soit attribué plutôt que d’abandonner les demandes. Après les avoir rejetés, il renvoie false, nous pouvons donc les mettre en file d’attente si nous le voulons vraiment.

if (limiter.enter()) {
System.out.println("Values:- " + Thread.currentThread().getName() + ": " + i);
} else { // queue the work again }

Lorsque nous essayons de régler le TPS sur 2 nous voyons la sortie suivante :

Ça marche!

1. Considérez un cas où vous écrivez du code pour capturer la signature de l’utilisateur. Pendant qu’ils font glisser le pointeur, vous capturez des milliers de points. Tous ne sont peut-être pas nécessaires pour une signature fluide, vous prenez donc un échantillon en utilisant la limitation de débit.
2. Certains événements sont appelés à une fréquence élevée. Vous pouvez contrôler cela.
3. Nous avons MessageQueueest l’auditeur oisif. Lorsque nous l’écoutons dans le fil principal, il est appelé de manière aléatoire. Parfois, il est appelé plusieurs fois en une seconde. Si nous voulons construire un système de battement de cœur qui nous indique quand le thread principal est inactif, nous pouvons l’utiliser pour recevoir des événements par seconde. Si nous ne recevons pas d’événement pendant une seconde, nous pouvons supposer que le thread principal est occupé.
4. Pour la gestion des quotas d’API de votre framework/bibliothèque, vous pouvez contrôler les appels d’API en fonction du plan de paiement choisi par l’utilisateur.

C’est tout.

Nous construirons un limiteur de débit plus complexe dans les prochains articles.