Как реализовать двусторонний обмен данными между потоками

Двусторонний обмен между потоками

Общая переменная передает состояние, но не описывает протокол: кто пишет, кто ждет, где конец и что делать при переполнении. Для обмена сообщениями лучше явно задать два направления двумя BlockingQueue.

record Request(long id, String payload) {}
record Response(long id, String payload) {}
 
BlockingQueue<Request> requests = new ArrayBlockingQueue<>(100);
BlockingQueue<Response> responses = new ArrayBlockingQueue<>(100);
 
Thread server = new Thread(() -> {
    try {
        while (!Thread.currentThread().isInterrupted()) {
            Request request = requests.take();
            responses.put(handle(request));
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});
 
server.start();
requests.put(new Request(1, "ping"));
Response response = responses.take();

Ограниченная очередь дает backpressure: put() ждет свободное место. Для запроса-ответа в реальной системе полезно сопоставлять сообщения по id, вводить deadline и передавать сигнал завершения.

Другие варианты:

• SynchronousQueue — передача из рук в руки без буфера;
• Exchanger<T> — парный обмен значениями в одной точке;
• CompletableFuture<Response> внутри запроса — адресный ответ;
• PipedInputStream/PipedOutputStream — поток байтов, но сложнее отмена и диагностика.

Атомарные переменные подходят для простого последнего состояния, но не заменяют очередь сообщений.

На собеседовании

• Зачем две очереди? Они явно моделируют два независимых направления.
• Почему очередь лучше общего List? Она атомарно координирует ожидание и видимость.
• Что дает bounded queue? Ограничение памяти и обратное давление.
• Как завершить потребителя? Interrupt, закрываемый канал или специальное terminal-сообщение по согласованному протоколу.