java - Java API调用将根据队列大小进行合并，或者在最早的项目插入队列后的4秒内执行

Question

我找不到基于队列大小来合并API调用的最佳方法，也无法在将最老的项插入队列后的4秒内执行。请在下面的问题说明中查找:
问题陈述:
为了防止查询调用使API过载，我想合并每个API端点的调用。
每个单独API的所有传入请求都应保留在队列中，并按以下顺序转发给API:
一旦达到了针对单个API的调用上限(4个)。如果达到了特定API的上限，则将使用带有逗号分隔值的q参数发送单个请求。因此，如果未达到特定服务的队列上限，则调用者将不会收到对其请求的响应。为了解决这个问题，我希望在将最老的项目插入队列后的4秒内发送服务队列。

Example :

GET http://<host>:8080/products?q=190763,190764
200 OK
content-type: application/json
{
"190763": 140.00
"190764": 250.00
}

If there is a caller querying API and the queue of the Products API holds 4 
requests, the next request to the Products API will trigger the actual bulk 
request to be made. Each API will have its own queue.

我正在使用Spring Boot。我尝试将CompletableFuture与执行程序服务一起使用，但无法解决此问题。任何帮助，将不胜感激。

Answer 1

您的执行人服务方式有效。我尝试了一个简单的示例，其中包含3个任务和一个预定的执行程序服务:

任务创建传入请求

任务检查队列中是否有4个以上的请求，并调用API

任务检查等待时间是否为4秒，并调用API

对API的调用始终会合并队列中的请求。队列必须是线程安全的。

public class ConsolidateApiCalls {

public static final ConcurrentLinkedDeque<String> queue = new ConcurrentLinkedDeque<>();
public static LocalTime timeAdded = LocalTime.now();

public static void main(String[] args) throws JSONException, ExecutionException, InterruptedException {
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
    executor.scheduleAtFixedRate(() -> {
                queue.add("NL");
                if (queue.size() == 1) timeAdded = LocalTime.now();
            },
            1, 500, TimeUnit.MILLISECONDS);
    executor.scheduleAtFixedRate(() -> {
                if (queue.size() >= 4) {
                    callAPI("size");
                }
            },
            1, 200, TimeUnit.MILLISECONDS);
    executor.scheduleAtFixedRate(() -> {
                if (Duration.between(LocalTime.now(), timeAdded.plusSeconds(4)).isNegative() && !queue.isEmpty()) {
                    callAPI("time");
                }
            },
            1, 500, TimeUnit.MILLISECONDS);
}

public static void callAPI(String reason) {
    System.out.println("API called due to " + reason + " limit.");
    StringBuilder result = new StringBuilder();
    queue.forEach(s -> result.append(s).append(": ").append(ThreadLocalRandom.current().nextDouble()).append(", "));
    queue.clear();
    System.out.println(result);
}}

您可以更改每个任务的期限，并查看由于大小和时间限制而调用API的频率。