RateLimiter 限流处理

Myth丶恋晨 2021-10-19 21:56 391阅读 0赞

RateLimiter 从概念上来讲，速率限制器会在可配置的速率下分配许可证。如果必要的话，每个acquire() 会阻塞当前线程直到许可证可用后获取该许可证。一旦获取到许可证，不需要再释放许可证。

*校对注：RateLimiter使用的是一种叫令牌桶的流控算法，RateLimiter会按照一定的频率往桶里扔令牌，线程拿到令牌才能执行，比如你希望自己的应用程序QPS不要超过1000，那么RateLimiter设置1000的速率后，就会每秒往桶里扔1000个令牌。*

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>com.google.common.util.concurrent.RateLimiter</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>3</code></td> 
   <td><code>@ThreadSafe</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>4</code></td> 
   <td><code>@Betapublic</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>5</code></td> 
   <td><code>abstract</code>&nbsp;<code>class</code>&nbsp;<code>RateLimiter&nbsp;</code><code>extends</code>&nbsp;<code>Object</code></td> 
  </tr> 
 </tbody> 
</table>

RateLimiter经常用于限制对一些物理资源或者逻辑资源的访问速率。与Semaphore 相比，Semaphore 限制了并发访问的数量而不是使用速率。（注意尽管并发性和速率是紧密相关的，比如参考[Little定律][Little]）

通过设置许可证的速率来定义RateLimiter。在默认配置下，许可证会在固定的速率下被分配，速率单位是每秒多少个许可证。为了确保维护配置的速率，许可会被平稳地分配，许可之间的延迟会做调整。  
可能存在配置一个拥有预热期的RateLimiter 的情况，在这段时间内，每秒分配的许可数会稳定地增长直到达到稳定的速率。

举例来说明如何使用RateLimiter，想象下我们需要处理一个任务列表，但我们不希望每秒的任务提交超过两个：

//速率是每秒两个许可
    final RateLimiter rateLimiter = RateLimiter.create(2.0);
    
    void submitTasks(List tasks, Executor executor) {
        for (Runnable task : tasks) {
            rateLimiter.acquire(); // 也许需要等待
            executor.execute(task);
        }
    }

**Since:**13.0  **作者:   **Dimitris Andreou再举另外一个例子，想象下我们制造了一个数据流，并希望以每秒5kb的速率处理它。可以通过要求每个字节代表一个许可，然后指定每秒5000个许可来完成：

// 每秒5000个许可
    final RateLimiter rateLimiter = RateLimiter.create(5000.0); 
    
    void submitPacket(byte[] packet) {
        rateLimiter.acquire(packet.length);
        networkService.send(packet);
    }

<table> 
 <tbody> 
  <tr> 
   <th><strong>setRate</strong></th> 
  </tr> 
  <tr> 
   <td>public final&nbsp;void&nbsp;setRate(double&nbsp;permitsPerSecond)<br> 更新RateLimite的稳定速率，参数permitsPerSecond&nbsp;由构造RateLimiter的工厂方法提供。调用该方法后，当前限制线程不会被唤醒，因此他们不会注意到最新的速率；只有接下来的请求才会。需要注意的是，由于每次请求偿还了（通过等待，如果需要的话）上一次请求的开销，这意味着紧紧跟着的下一个请求不会被最新的速率影响到，在调用了setRate&nbsp;之后；它会偿还上一次请求的开销，这个开销依赖于之前的速率。RateLimiter的行为在任何方式下都不会被改变，比如如果&nbsp;RateLimiter&nbsp;有20秒的预热期配置，在此方法被调用后它还是会进行20秒的预热。<br><strong>参数:</strong><br> permitsPerSecond&nbsp;– RateLimiter的新的稳定速率<br><strong>抛出:</strong><br> IllegalArgumentException&nbsp;– 如果permitsPerSecond为负数或者为0</td> 
  </tr> 
 </tbody> 
</table>

有一点很重要，那就是请求的许可数从来不会影响到请求本身的限制（调用acquire(1) 和调用acquire(1000) 将得到相同的限制效果，如果存在这样的调用的话），但会影响下一次请求的限制，也就是说，如果一个高开销的任务抵达一个空闲的RateLimiter，它会被马上许可，但是下一个请求会经历额外的限制，从而来偿付高开销任务。注意：RateLimiter 并不提供公平性的保证。

## 方法摘要 ##

<table> 
 <tbody> 
  <tr> 
   <th><strong>修饰符和类型</strong></th> 
   <th><strong>方法和描述</strong></th> 
  </tr> 
  <tr> 
   <td>double</td> 
   <td>acquire()<br> 从RateLimiter获取一个许可，该方法会被阻塞直到获取到请求</td> 
  </tr> 
  <tr> 
   <td>double</td> 
   <td>acquire(int&nbsp;permits)<br> 从RateLimiter获取指定许可数，该方法会被阻塞直到获取到请求</td> 
  </tr> 
  <tr> 
   <td>static&nbsp;RateLimiter</td> 
   <td>create(double&nbsp;permitsPerSecond)<br> 根据指定的稳定吞吐率创建RateLimiter，这里的吞吐率是指每秒多少许可数（通常是指QPS，每秒多少查询）</td> 
  </tr> 
  <tr> 
   <td>static&nbsp;RateLimiter</td> 
   <td>create(double&nbsp;permitsPerSecond, long&nbsp;warmupPeriod,&nbsp;TimeUnit&nbsp;unit)<br> 根据指定的稳定吞吐率和预热期来创建RateLimiter，这里的吞吐率是指每秒多少许可数（通常是指QPS，每秒多少个请求量），在这段预热时间内，RateLimiter每秒分配的许可数会平稳地增长直到预热期结束时达到其最大速率。（只要存在足够请求数来使其饱和）</td> 
  </tr> 
  <tr> 
   <td>double</td> 
   <td>getRate()<br> 返回RateLimiter&nbsp;配置中的稳定速率，该速率单位是每秒多少许可数</td> 
  </tr> 
  <tr> 
   <td>void</td> 
   <td>setRate(double&nbsp;permitsPerSecond)<br> 更新RateLimite的稳定速率，参数permitsPerSecond&nbsp;由构造RateLimiter的工厂方法提供。</td> 
  </tr> 
  <tr> 
   <td>String</td> 
   <td>toString()<br> 返回对象的字符表现形式</td> 
  </tr> 
  <tr> 
   <td>boolean</td> 
   <td>tryAcquire()<br> 从RateLimiter&nbsp;获取许可，如果该许可可以在无延迟下的情况下立即获取得到的话</td> 
  </tr> 
  <tr> 
   <td>boolean</td> 
   <td>tryAcquire(int&nbsp;permits)<br> 从RateLimiter&nbsp;获取许可数，如果该许可数可以在无延迟下的情况下立即获取得到的话</td> 
  </tr> 
  <tr> 
   <td>boolean</td> 
   <td>tryAcquire(int&nbsp;permits, long&nbsp;timeout,&nbsp;TimeUnit&nbsp;unit)<br> 从RateLimiter&nbsp;获取指定许可数如果该许可数可以在不超过timeout的时间内获取得到的话，或者如果无法在timeout 过期之前获取得到许可数的话，那么立即返回false&nbsp;（无需等待）</td> 
  </tr> 
  <tr> 
   <td>boolean</td> 
   <td>tryAcquire(long&nbsp;timeout,&nbsp;TimeUnit&nbsp;unit)<br> 从RateLimiter&nbsp;获取许可如果该许可可以在不超过timeout的时间内获取得到的话，或者如果无法在timeout 过期之前获取得到许可的话，那么立即返回false（无需等待）</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th>从java.lang.Object 类继承的方法</th> 
  </tr> 
  <tr> 
   <td>clone,&nbsp;equals,&nbsp;finalize,&nbsp;getClass,&nbsp;hashCode,&nbsp;notify,&nbsp;notifyAll,&nbsp;wait,&nbsp;wait,&nbsp;wait</td> 
  </tr> 
 </tbody> 
</table>

## 方法细节 ##

<table> 
 <tbody> 
  <tr> 
   <th><strong>create</strong></th> 
  </tr> 
  <tr> 
   <td>public static&nbsp;RateLimiter&nbsp;create(double&nbsp;permitsPerSecond)<br> 根据指定的稳定吞吐率创建RateLimiter，这里的吞吐率是指每秒多少许可数（通常是指QPS，每秒多少查询）。<br> The returned&nbsp;RateLimiter&nbsp;ensures that on average no more than&nbsp;permitsPerSecond&nbsp;are issued during any given second, with sustained requests being smoothly spread over each second. When the incoming request rate exceeds&nbsp;permitsPerSecond&nbsp;the rate limiter will release one permit every&nbsp;(1.0 / permitsPerSecond)&nbsp;seconds. When the rate limiter is unused, bursts of up to&nbsp;permitsPerSecond&nbsp;permits will be allowed, with subsequent requests being smoothly limited at the stable rate of&nbsp;permitsPerSecond.<br> 返回的RateLimiter&nbsp;确保了在平均情况下，每秒发布的许可数不会超过permitsPerSecond，每秒钟会持续发送请求。当传入请求速率超过permitsPerSecond，速率限制器会每秒释放一个许可(1.0 / permitsPerSecond 这里是指设定了permitsPerSecond为1.0)&nbsp;。当速率限制器闲置时，允许许可数暴增到permitsPerSecond，随后的请求会被平滑地限制在稳定速率permitsPerSecond中。<br><strong>参数:</strong><br> permitsPerSecond&nbsp;– 返回的RateLimiter的速率，意味着每秒有多少个许可变成有效。<br><strong>抛出:</strong><br> IllegalArgumentException&nbsp;– 如果permitsPerSecond为负数或者为0</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>create</strong></th> 
  </tr> 
  <tr> 
   <td>public static&nbsp;RateLimiter&nbsp;create(double&nbsp;permitsPerSecond,long&nbsp;warmupPeriod,TimeUnit&nbsp;unit)<br> 根据指定的稳定吞吐率和预热期来创建RateLimiter，这里的吞吐率是指每秒多少许可数（通常是指QPS，每秒多少查询），在这段预热时间内，RateLimiter每秒分配的许可数会平稳地增长直到预热期结束时达到其最大速率（只要存在足够请求数来使其饱和）。同样地，如果RateLimiter&nbsp;在warmupPeriod时间内闲置不用，它将会逐步地返回冷却状态。也就是说，它会像它第一次被创建般经历同样的预热期。返回的RateLimiter&nbsp;主要用于那些需要预热期的资源，这些资源实际上满足了请求（比如一个远程服务），而不是在稳定（最大）的速率下可以立即被访问的资源。返回的RateLimiter&nbsp;在冷却状态下启动（即预热期将会紧跟着发生），并且如果被长期闲置不用，它将回到冷却状态。<br><strong>参数:</strong> <p>&nbsp;</p> 
    <ul> 
     <li>permitsPerSecond&nbsp;– 返回的RateLimiter的速率，意味着每秒有多少个许可变成有效。</li> 
     <li>warmupPeriod&nbsp;– 在这段时间内RateLimiter会增加它的速率，在抵达它的稳定速率或者最大速率之前</li> 
     <li>unit&nbsp;– 参数warmupPeriod 的时间单位</li> 
    </ul><p><strong>抛出:</strong></p> 
    <ul> 
     <li>IllegalArgumentException –&nbsp;如果permitsPerSecond为负数或者为0</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>getRate</strong></th> 
  </tr> 
  <tr> 
   <td>public final&nbsp;double&nbsp;getRate()<br> 返回RateLimiter&nbsp;配置中的稳定速率，该速率单位是每秒多少许可数。它的初始值相当于构造这个RateLimiter的工厂方法中的参数permitsPerSecond&nbsp;，并且只有在调用setRate(double)后才会被更新。</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>acquire</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;double&nbsp;acquire()<br> 从RateLimiter获取一个许可，该方法会被阻塞直到获取到请求。如果存在等待的情况的话，告诉调用者获取到该请求所需要的睡眠时间。该方法等同于acquire(1)。<br><strong>返回:</strong><br> time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited<br> 执行速率的所需要的睡眠时间，单位为妙；如果没有则返回0<br><strong>Since:</strong><br> 16.0 (版本13.0没有返回值)</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>acquire</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;double&nbsp;acquire(int&nbsp;permits)<br> 从RateLimiter获取指定许可数，该方法会被阻塞直到获取到请求数。如果存在等待的情况的话，告诉调用者获取到这些请求数所需要的睡眠时间。<br><strong>参数:</strong><br> permits&nbsp;– 需要获取的许可数<br><strong>返回:</strong><br> 执行速率的所需要的睡眠时间，单位为妙；如果没有则返回0<br><strong>抛出:</strong><br> IllegalArgumentException&nbsp;– 如果请求的许可数为负数或者为0<br><strong>Since:</strong><br> 16.0 (版本13.0没有返回值)</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>tryAcquire</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;boolean&nbsp;tryAcquire(long&nbsp;timeout,TimeUnit&nbsp;unit)<br> 从RateLimiter获取许可如果该许可可以在不超过timeout的时间内获取得到的话，或者如果无法在timeout 过期之前获取得到许可的话，那么立即返回false（无需等待）。该方法等同于tryAcquire(1, timeout, unit)。<br><strong>参数:</strong> <p>&nbsp;</p> 
    <ul> 
     <li>timeout&nbsp;– 等待许可的最大时间，负数以0处理</li> 
     <li>unit&nbsp;– 参数timeout 的时间单位</li> 
    </ul><p><strong>返回:</strong><br> true表示获取到许可，反之则是false<br><strong>抛出:</strong><br> IllegalArgumentException&nbsp;– 如果请求的许可数为负数或者为0</p> </td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>tryAcquire</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;boolean&nbsp;tryAcquire(int&nbsp;permits)<br> 从RateLimiter&nbsp;获取许可数，如果该许可数可以在无延迟下的情况下立即获取得到的话。该方法等同于tryAcquire(permits, 0, anyUnit)。<br><strong>参数:</strong><br> permits&nbsp;– 需要获取的许可数<br><strong>返回:</strong><br> true表示获取到许可，反之则是false<br><strong>抛出:</strong><br> IllegalArgumentException&nbsp;– 如果请求的许可数为负数或者为0<br><strong>Since:</strong><br> 14.0</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>tryAcquire</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;boolean&nbsp;tryAcquire()<br> 从RateLimiter&nbsp;获取许可，如果该许可可以在无延迟下的情况下立即获取得到的话。<br> 该方法等同于tryAcquire(1)。<br><strong>返回:</strong><br> true表示获取到许可，反之则是false<br><strong>Since:</strong><br> 14.0</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>tryAcquire</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;boolean&nbsp;tryAcquire(int&nbsp;permits,long&nbsp;timeout,TimeUnit&nbsp;unit)<br> 从RateLimiter&nbsp;获取指定许可数如果该许可数可以在不超过timeout的时间内获取得到的话，或者如果无法在timeout 过期之前获取得到许可数的话，那么立即返回false&nbsp;（无需等待）。<br><strong>参数:</strong><br> permits&nbsp;– 需要获取的许可数<br> timeout&nbsp;– 等待许可数的最大时间，负数以0处理<br> unit&nbsp;– 参数timeout 的时间单位<br><strong>返回:</strong><br> true表示获取到许可，反之则是false<br><strong>抛出:</strong><br> IllegalArgumentException&nbsp;-如果请求的许可数为负数或者为0</td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <th><strong>toString</strong></th> 
  </tr> 
  <tr> 
   <td>public&nbsp;String&nbsp;toString()<br> 以下描述复制于java.lang.Object类。<br> 返回对象的字符表现形式。通常来讲，toString&nbsp;方法返回一个“文本化呈现”对象的字符串。<br> 结果应该是一个简明但易于读懂的信息表达式。建议所有子类都重写该方法。<br> toString 方法返回一个由实例的类名，字符’@’和以无符号十六进制表示的对象的哈希值组成的字符串。换句话说，该方法返回的字符串等同于：<br> getClass().getName() + ‘@’ + Integer.toHexString(hashCode())<br><strong>重载:</strong><br> Object类的toString方法<br><strong>返回:</strong><br> 对象的字符表现形式</td> 
  </tr> 
 </tbody> 
</table>

[Little]: http://en.wikipedia.org/wiki/Little'%20rel=