如何限制一个时间段内的线程数答案

【问题标题】：How to limit number of threads within a time period如何限制一个时间段内的线程数
【发布时间】：2019-10-10 19:56:51
【问题描述】：

我正在使用的服务在 1 秒内发出 5 个请求后开始阻止请求。

在 Spring 中使用 Java 我正在寻找一种方法来对线程进行排队，这样一秒钟内最多 5 个线程可以访问临界区，并且一旦有带宽让它们继续，任何其他线程都会排队并释放.

目前我已经尝试使用锁进行此操作，但它会导致线程总是等待 1/5 秒，即使我们不会在不休眠的情况下达到每秒的最大调用次数。

    Lock l = new ReentrantLock();
    try {
        l.lock();
      //critical section
   } finally {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        l.unlock();
    }

有了这个实现，我永远不会超过每秒 5 次，但在一切准备好返回给用户之后，我也会导致响应延迟 200 毫秒。

我需要一个仅在需要延迟时才延迟线程的解决方案。在这种情况下，一秒钟内的第 6 个以上呼叫应该延迟，但前 5 个呼叫不需要延迟。同样，呼叫 6-11 可以同时进行。

【问题讨论】：

试试 Guava 的 RateLimiter。
为什么要限制每秒而不是一次只允许 5 个？
@daniu 当一个人完成另一个需要被允许进入。如果他们完成得非常快，超过 5 个可以在一秒钟内执行。上面的关键代码部分使用了第 3 方 api，每秒 5 次调用后将拒绝其他调用。
一旦你决定做什么，你需要释放锁。做实际工作时，你不能拿着锁。

标签： java multithreading concurrency critical-section

【解决方案1】：

这种速率限制是微服务架构中相当普遍的问题，因为它是解决级联故障的更广泛问题的一部分。周围有许多库来处理这个问题，其中一个使用最广泛的现代库称为Resilience4j，它提供了RateLimiter 实现。你可能想要一些非常接近这个的东西：

创建限制器：

RateLimiterConfig config = RateLimiterConfig.custom()
  .limitRefreshPeriod(Duration.ofSeconds(1))
  .limitForPeriod(5)
  .timeoutDuration(Duration.ofSeconds(4)) //or however long you want to wait before failing
  .build();

// Create registry
RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(config);

// Use registry
RateLimiter rateLimiter = rateLimiterRegistry
  .rateLimiter("someServiceLimiter", config);

使用它：

// Decorate your call to BackendService.doSomething()
CheckedRunnable restrictedCall = RateLimiter
    .decorateCheckedRunnable(rateLimiter, backendService::doSomething);

//Or, you can use an annotation:
@RateLimiter(name = "someServiceLimiter")
public void doSomething() {
    //backend call
}

【讨论】：

我添加了依赖项，但是对于@RateLimiter，我收到一个错误，指出它不是有效的注释。知道我错过了什么吗？
注解RateLimiter和实现RateLimiter同名。因此，如果您在同一个文件中同时使用两者，则需要指定其中一个的整个包。例如@io.github.resilience4j.ratelimiter.annotation.RateLimiter
谢谢。我能够编译它，但仍然能够让代码执行超过limitForPeriod。我在包含关键部分的同一个类中有这个逻辑。配置应该在其他地方吗？我在文档中看到他们把它放在 yaml 中，但那不是为我编译的
你在使用io.github.resilience4j:resilience4j-spring-boot2 依赖吗？如果您仍然遇到问题，您可能希望使用不带注释的方法。

【解决方案2】：

我认为使用semaphore API 解决它是最好的方法。

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class BulkheadSemaphore {

    private Queue<Long> enterQueue = new LinkedList<>();
    private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    private Semaphore semaphore;

    public BulkheadSemaphore(final Long timeLimit, final int concurrentThreadsLimit) {
        this.semaphore = new Semaphore(concurrentThreadsLimit);

        executor.scheduleAtFixedRate(() -> {
            final Long now = now();

            while (!enterQueue.isEmpty() && now - enterQueue.peek() >= timeLimit) {
                enterQueue.poll();
                semaphore.release();
            }
        }, timeLimit, 200, TimeUnit.MILLISECONDS);
    }

    private Long now() {
        return System.currentTimeMillis();
    }

    public void acquire() {
        try {
            semaphore.acquire();
        } catch (InterruptedException e) {
            // todo: handle exception
        }
    }

    public void release() {
        semaphore.release();
    }
}

api很简单：

每个线程进入临界区，调用bulkheadSemaphore.acqure()
外部调用执行完成后，调用bulkheadSemaphore.release()

为什么它可以解决问题？

此信号量为进入很久以前的关键部分。
它以一定的速率释放它的许可（我将它设置为 200 毫秒，不过它可以更小）。它还保证如果一个工作单元已经快速完成，下一个线程将能够启动一个新的工作单元。
有些线程仍会面临冗余等待，但并非每次都发生，它们最多会花费 200 毫秒。

由于请求需要时间，我将 timeLimit 设置为 1.5 秒以匹配您的 1 秒限制。

附：不要忘记关闭执行器服务

【讨论】：