Как добавить пакетную обработку, неявную для клиента?

Если заменить возвращаемый тип requestRowBatch с List<String> на Map<Integer, String>, чтобы делегировать сопоставление запроса и ответа серверу, я смог создать следующее решение:

@Slf4j
public class MyClientSideServiceCompletableFuture {
    private final Integer batchSize = 25;
    private final Integer timeoutMillis = 5000;
    private final MyServerSideService myServerSideService;
    private final BlockingQueue<Pair<Integer, CompletableFuture>> queue = new LinkedBlockingQueue<>();

    private final Lock lock = new ReentrantLock();
    private final Condition requestAddedCondition = lock.newCondition();


    public MyClientSideServiceCompletableFuture(MyServerSideService myServerSideService) {
        this.myServerSideService = myServerSideService;
        startQueueDrainer();
    }

    public String requestSingleRow(int req) {
        CompletableFuture<String> future = new CompletableFuture<>();
        while (!queue.offer(Pair.of(req, future))) {
            log.error("Can't add {} to the queue. Retrying...", req);
        }
        lock.lock();
        try {
            requestAddedCondition.signal();
        } finally {
            lock.unlock();
        }
        try {
            return future.get();
        } catch (Exception e) {
            return "Exception";
        }
    }

    private void startQueueDrainer() {
        new Thread(() -> {
            log.info("request");
            while (true) {
                ArrayList<Pair<Integer, CompletableFuture>> requests = new ArrayList<>();
                if (queue.drainTo(requests, batchSize) > 0) {
                    log.info("drained {} items", requests.size());
                    Map<Integer, String> serverResponseMap = requestBatch(requests.stream().map(Pair::getKey).collect(Collectors.toList()));
                    requests.forEach(pair -> {
                        String response = serverResponseMap.get(pair.getKey());
                        CompletableFuture<String> value = pair.getValue();
                        value.complete(response);
                    });
                } else {
                    lock.lock();
                    try {
                        while (queue.size() == 0) {
                            try {
                                log.info("Waiting on condition");
                                requestAddedCondition.await(timeoutMillis, TimeUnit.MILLISECONDS);
                                log.info("Waking up on condition");
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    } finally {
                        lock.unlock();
                    }
                }

            }
        }).start();
    }


    public Map<Integer, String> requestBatch(Collection<Integer> requestList) {
        return myServerSideService.requestRowBatch(requestList);
    }
}

Похоже на рабочее решение. Но я не уверен, что это оптимально.

Ваше решение MyClientSideServiceCompletableFuture будет отправлять запросы на сервер каждый раз, когда вы добавляете что-то в очередь, и не ждет, пока запросы будут иметь пакетный размер. Вы используете BlockingQueue и добавляете ненужное условие блокировки и блокировки. BlockingQueue имеет возможности блокировки тайм-аута, поэтому никаких дополнительных условий не требуется.

Вы можете упростить свое решение следующим образом:

Он отправляет запросы на сервер только тогда, когда пакет заполнен или истек тайм-аут, а пакет не пуст.

private void startQueueDrainer() {

        new Thread(() -> {
            log.info("request");
            ArrayList<Pair<Integer, CompletableFuture>> batch = new ArrayList<>(batchSize);
            while (true) {
                try {
                    batch.clear(); //clear batch
                    long timeTowWait = timeoutMillis;
                    long startTime = System.currentTimeMillis();

                    while (timeTowWait > 0 && batch.size() < batchSize) {
                        Pair<Integer, CompletableFuture> request = queue.poll(timeTowWait , TimeUnit.MILLISECONDS);
                        if(request != null){
                          batch.add(request);
                        }
                        long timeSpent = (System.currentTimeMillis() - startTime);
                        timeTowWait = timeTowWait - timeSpent;
                    }

                    if (!batch.isEmpty()) {
                        // we wait at least timeoutMillis or batch is full
                        log.info("send {} requests to server", batch.size());
                        Map<Integer, String> serverResponseMap = requestBatch(batch.stream().map(Pair::getKey).collect(Collectors.toList()));
                        batch.forEach(pair -> {
                            String response = serverResponseMap.get(pair.getKey());
                            CompletableFuture<String> value = pair.getValue();
                            value.complete(response);
                        });
                    } else {
                        log.info("We wait {} but the batch is still empty", System.currentTimeMillis() - startTime);
                    }

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

Измените метод requestSingleRow, чтобы он не использовал блокировку.

  public String requestSingleRow(int req) {
        CompletableFuture<String> future = new CompletableFuture<>();

        while (!queue.offer(Pair.of(req, future))) {
            log.error("Can't add {} to the queue. Retrying...", req);
        }

        try {
            return future.get();
        } catch (Exception e) {
            return "Exception";
        }
    }

Вы можете использовать CompletableFuture.
Попросите потоки, вызывающие MyClientSideService, поместить свой запрос в Queue (возможно, BlockingQueue, и получить взамен новый CompletableFuture. Вызывающий поток может вызвать CompletableFuture.get() для блокировки до тех пор, пока не будет готов результат, или продолжить выполнение других действий. .

Этот CompletableFuture будет храниться вместе с запросом в MyClientSideService. Когда вы достигнете 50 запросов (и, следовательно, 50 CompletableFuture экземпляров), попросите клиентскую службу отправить пакетный запрос.

По завершении запроса используйте метод CompletableFuture.complete(value) каждого экземпляра ComplatableFuture в очереди, чтобы уведомить поток клиента о готовности ответа. Это разблокирует клиент, если он вызвал метод блокировки, такой как CompletableFuture.get(), или заставит его немедленно вернуться со значением, если он будет вызван позже.