Java вызовов API, которые должны быть объединены в зависимости от размера очереди или выполнены в течение 4 секунд после того, как самый старый элемент был вставлен в очередь

Question

Я не могу найти лучший подход для объединения вызовов API на основе размера очереди или выполнения в течение 4 секунд после того, как самый старый элемент был вставлен в очередь. Ниже приведено описание проблемы:

Описание проблемы:

Чтобы предотвратить перегрузку API-интерфейсов вызовами запросов, я хотел бы объединить вызовы для каждой конечной точки API.

Все входящие запросы для каждого отдельного API должны храниться в очереди и пересылаться в API, как только будет достигнуто ограничение в 4 вызова для отдельного API. Если достигнуто ограничение для определенного c API, будет отправлен один запрос с использованием параметра q со значениями, разделенными запятыми. Из-за этого вызывающий абонент не получит ответа на свои запросы, если не будет достигнут предел очереди для определенной службы c. Чтобы решить эту проблему, я хочу, чтобы очереди обслуживания также отправлялись в течение 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 из-за размера и ограничения по времени.