怎样用 Java 来构建一个简单的速率限制器？

速率限制

现实世界中的用户是残暴的，而且没耐烦，布满着各种不确定性。在高并发体系中，可能会出现服务器被虚伪哀求轰炸的情况，因此您可能渴望控制这种情况。
一些实际使用情况可能如下所示：

• API配额管理-作为提供者，您可能渴望根据用户的付款情况限制向服务器发出API哀求的速率。这可以在客户端或服务端实现。
  
• 安全性-防止DDOS攻击。
  
• 成本控制--这对服务方乃至客户方来说都不是必须的。如果某个组件以非常高的速率发出一个事件，它可能有助于控制它，它可能有助于控制从客户端发送的遥测。
  

限速处理时的选项

根据我们处理的哀求/事件类型，可能会发生以下情况：

• 我们可以放弃额外的哀求
  
• 我们可以选择让哀求等待，直到体系将它们降低到预定义的速率。
  

常用限速算法

• 令牌桶算法
  
• 漏桶算法
  

我们将不深入讨论这些算法的内部细节，由于这超出了本文的范围。
我们将以令牌桶算法为中央。其要求如下。
令牌桶算法基于以固定速率添加令牌的固定容量桶的类比。在答应API继续之前，将查抄桶，以检察它当时是否包罗至少一个令牌。如果令牌存在，则进行API调用。如果不是，则丢弃该消息/或使其等待。
需求

• 应该可以大概接受每秒所需的（TPS）事务或速率。
  
• 如果超过我们定义的比率，则应放弃交易。
  
• 应该在同时发生的情况下起作用。
  

高级功能（在后续文章中实现）

• 应该可以大概平滑突发的哀求。例如，如果我们将TPS定义为5，而且所有五个哀求都在同一时间到达，那么它应该可以大概以固定的时间间隔将它们排成一行，即以200ms的时间间隔实行每个哀求。它需要一个内部定时电路。
  
• 如果我们的TPS为5，而且在其中一个1秒的时段中，我们在下一秒只使用3个代币，那么我们应该可以大概提供5+2 = 7个代币作为嘉奖。但速率为每个令牌1/7（142.28ms）。奖金不应结转到下一个插槽。
  

让我们起首定义我们的 速率限制器：

1. /**
2. * Rate limiter helps in limiting the rate of execution of a piece of code. The rate is defined in terms of
3. * TPS(Transactions per second). Rate of 5 would suggest, 5 transactions/second. Transaction could be a DB call, API call,
4. * or a simple function call.
5. * <p>
6. * Every {@link RateLimiter} implementation should implement either {@link RateLimiter#throttle(Code)} or, {@link RateLimiter#enter()}.
7. * They can also choose to implement all.
8. * <p>
9. * {@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
10. * spreads across multiple functions, we need to use entry() and exit() contract.
11. */
12. public interface RateLimiter {
14. /**
15.      * Rate limits the code passed inside as an argument.
16.      *
17.      * @param code representation of the piece of code that needs to be rate limited.
18.      * @return true if executed, false otherwise.
19.      */
20.     boolean throttle(Code code);
21.     /**
22.      * When the piece of code that needs to be rate limited cannot be represented as a contiguous
23.      * code, then entry() should be used before we start executing the code. This brings the code inside the rate
24.      * limiting boundaries.
25.      *
26.      * @return true if the code will execute and false if rate limited.
27.      * <p
28.      */
29.     boolean enter();
30.     /**
31.      * Interface to represent a contiguous piece of code that needs to be rate limited.
32.      */
33.     interface Code {
34.         /**
35.          * Calling this function should execute the code that is delegated to this interface.
36.          */
37.         void invoke();
38.     }
39. }
40. 复制代码

复制代码

我们的 RateLimit有两组API：一个是throttle（code），另一个是enter（）。这两种方法都满足相同的功能，但采用以下两种方式：

• boolean throttle（代码）-如果我们有连续的代码，可以用来通报一个代码块。
  
• 布尔输入（） - 通常可以在API、DB或任何我们想要节省的调用之前使用。如果实行此代码后面的代码，则将返回 真 ，以及 假的如果它是速率受限的话。您可以将这些哀求排队或拒绝。
  

   在生产环境中您永世不会看到节省（代码）实现，由于它不是最佳的。请在评论中告诉我原因。大多数速率限制器使用雷同于enter（）的API。
  核心功能

为了构建速率限制器的核心，我们需要确保在恣意两秒之间不答应超过N个事务。我们将怎样做到这一点？
考虑我们进行第一笔交易的时间t0。 t0 .所以，
直到（t0 + 1）s，我们只答应进行N次交易。 (t0 + 1)s , we are allowed to make only N transactions.怎样确保这一点？在下次交易时，我们将查抄
当前时间≤（t0 + 1）。.如果没有，那么这意味着我们进入了不同的秒，而且我们被答应进行N次交易。 N transactions.让我们看一小段代码，它演示了：

1. long now = System.nanoTime();
2. if (now <= mNextSecondBoundary) { // If we are within the time limit of current second
3.     if (mCounter < N) { // If token available
4.         mLastExecutionNanos = now;
5.         mCounter++; // Allocate token
6.         invoke(code); // Invoke the code passed the throttle method.
7.     }
8. }
9. 复制代码

复制代码

那么，我们怎样定义mNextSecondBoundary呢？这将在我们进行第一个事务时完成，如前所述，我们将在完成第一个事务的时间增加一秒。

1. if (mLastExecutionNanos == 0L) {
2.     mCounter++; // Allocate the very first token here.
3.     mLastExecutionNanos = System.nanoTime();
4.     mNextSecondBoundary = mLastExecutionNanos + NANO_PER_SEC;  // (10^9)
5. }
6. 复制代码

复制代码

如今，如果我们实行代码并看到我们进入了不同的秒，我们应该怎么做？我们将通过重置上次实行时间、可用令牌数来增强前面的代码，并通过调用 节省阀（）再一次。我们的方法已经知道怎样处理新的秒。

1. @Override
2. public boolean throttle(Code code) {
3.     if (mTPS <= 0) {
4.         // We do not want anything to pass.
5.         return false;
6.     }
8. synchronized (mLock) {
9.         if (mLastExecutionNanos == 0L) {
10.             mCounter++;
11.             mLastExecutionNanos = System.nanoTime();
12.             mNextSecondBoundary = mLastExecutionNanos + NANO_PER_SEC;
13.             invoke(code);
14.             return true;
15.         } else {
16.             long now = System.nanoTime();
17.             if (now <= mNextSecondBoundary) {
18.                 if (mCounter < mTPS) {
19.                     mLastExecutionNanos = now;
20.                     mCounter++;
21.                     invoke(code);
22.                     return true;
23.                 } else {
24.                     return false;
25.                 }
26.             } else {
27.                 // Reset the counter as we in a different second now.
28.                 mCounter = 0;
29.                 mLastExecutionNanos = 0L;
30.                 mNextSecondBoundary = 0L;
31.                 return throttle(code);
32.             }
33.         }
34.     }
35. }
36. 复制代码

复制代码

在这个实现中，我们可以通报需要节省的代码块，但是这个代码有一个问题。这将工作，但它会表现不佳。不保举，但为什么呢？请在评论中告诉我。
如今，可以使用相同的构建块和enter（）构建第二个API了。我们将使用相同的逻辑，但我们不会实行方法内部的代码块。相反，它将在调用enter（）之后实行，就像我们实行状态管理一样。该方法的实现如下：

1. @Override
2. public boolean enter() {
3.     if (mTPS == 0L) {
4.         return false;
5.     }
7. synchronized (mBoundaryLock) {
8.         if (mLastExecutionNanos == 0L) {
9.             mLastExecutionNanos = System.nanoTime();
10.             mCounter++;
11.             mNextSecondBoundary = mLastExecutionNanos + NANO_PER_SEC;
12.             return true;
13.         } else {
14.             long now = System.nanoTime();
15.             if (now <= mNextSecondBoundary) {
16.                 if (mCounter < mTPS) {
17.                     mLastExecutionNanos = now;
18.                     mCounter++;
19.                     return true;
20.                 } else return false;
21.             } else {
22.                 // Reset the counter as we in a different second now.
23.                 mCounter = 0;
24.                 mLastExecutionNanos = 0L;
25.                 mNextSecondBoundary = 0L;
26.                 return enter();
27.             }
28.         }
29.     }
30. }
31. 复制代码

复制代码

如今，我们简单的速率限制器已经可以使用了。您可以检察完整的代码 这里。
结果

我们将尝试创建一个可创建六个线程的驱动步伐代码。每个线程尝试从0到100计数，延迟为50ms（可以设置为任何数字）。我们将按如下方式启动我们的限速器：

1. public static void main(String[] args) {
2.     RateLimiter limiter = new SimpleTokenBucketRateLimiter(1);
3.     Thread[] group = new Thread[6];
4.     Runnable r = () -> {
5.         for (int i = 0; i < 100; i++) {
6.             try {
7.                 Thread.sleep(50);
8.             } catch (InterruptedException e) {
9.                 throw new RuntimeException(e);
10.             }
11.             if (limiter.enter()) {
12.                 System.out.println("Values:- " + Thread.currentThread().getName() + ": " + i);
13.             }
14.         }
15.     };
18. for (int i = 0; i < 6; i++) {
19.         group[i] = new Thread(r);
20.         group[i].start();
21.     }
22. }
23. 复制代码

复制代码

我们的API不支持平滑事务，而是让事务等待下一个令牌被分配，而不是丢弃哀求。在拒绝它们之后，它返回false，所以如果我们真的想的话，我们可以把它们排队。

1. if (limiter.enter()) {
2.                 System.out.println("Values:- " + Thread.currentThread().getName() + ": " + i);
3. } else { // queue the work again }
4. 复制代码

复制代码

这是TPS设置为1时的输出。
当我们尝试将TPS设置为 2我们将看到以下输出：

真管用！
从Android的角度看

• 考虑这样一种情况：您正在编写代码以捕获用户签名。当他们拖动指针时，您会捕获数千个点。平滑签名可能不需要所有这些参数，因此您使用速率限制进行采样。
  
• 一些事件调用频率很高。你能控制的。
  
• 我们有MessageQueue的空闲侦听器。当我们在主线程中侦听它时，它被随意调用。偶然候，它在一秒钟内被调用好反复。如果我们想构建一个心跳体系来告诉我们主线程何时空闲，我们可以使用它来接收每秒的事件。如果我们一秒钟内没有收到事件，我们可以假定主线程处于忙碌状态。
  
• 对于您的框架/库的API配额管理，您可以根据用户选择的付款计划情况API调用。
  

今天先到这里吧。 我们将在后续文章中构建一个更复杂的速率限制器。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。