Реализуйте producer-consumer: один поток кладёт задачи, несколько обрабатывают. Какие инструменты java.util.concurrent возьмёте?
Короткий ответ
- BlockingQueue — готовое решение: put блокируется на полной, take на пустой
- Ограниченная ёмкость очереди даёт backpressure на производителя
- ArrayBlockingQueue или LinkedBlockingQueue с capacity
- Потребители — ExecutorService или явные потоки в цикле take
- Остановка: poison pill или interrupt + проверка InterruptedException
- Ручной wait/notify возможен, но ошибкоопасен — только если просят
Правильный ответ — ограниченная BlockingQueue с блокирующими put/take и аккуратной остановкой потребителей.
Как сказать вслух
пример ответаЯ возьму BlockingQueue с ограниченной ёмкостью. Производитель вызывает put — если очередь заполнена, он сам заблокируется, и это естественный backpressure. Потребители в цикле вызывают take, который ждёт появления элемента без всяких busy-wait. Останавливать потребителей можно «ядовитой пилюлей» — специальным объектом-маркером в очереди — или через interrupt. Писать это вручную на wait и notify я бы не стал без необходимости: в стандартной библиотеке всё уже есть и протестировано.
Подробный ответ
Основной ответ
Каркас: ArrayBlockingQueue<Task>(capacity) — ограниченная ёмкость обязательна, иначе при медленных потребителях очередь съест память. Производитель: queue.put(task) — блокируется на полной очереди (вариант offer с таймаутом, если нужно отбрасывать). Потребители: цикл Task t = queue.take(); process(t); запущенный в нескольких потоках (ExecutorService или виртуальные потоки). Остановка — два корректных способа: poison pill (по маркеру на потребителя; потребитель, получив его, выходит из цикла) или interrupt — take бросит InterruptedException, в обработчике восстановить флаг и выйти. Нужно проговорить гарантии: BlockingQueue потокобезопасна, happens-before между put и take обеспечивает видимость полей задачи. Если интервьюер просит «без библиотеки» — wait/notifyAll на мониторе с проверкой условия в цикле while (не if!).
Ключевые моменты
- Bounded queue. Ограниченная ёмкость = backpressure; безграничная очередь — отложенный OOM.
- put/take против offer/poll. Блокирующие версии для классической схемы; с таймаутами — когда нужна деградация.
- Остановка. Poison pill или interrupt; просто «убить» потоки нельзя — задачи потеряются.
- Ручная версия. wait в цикле while по условию, notifyAll после изменения; типичная ошибка — if вместо while.
Практический контекст
Задача проверяет практическое владение java.util.concurrent: кандидаты, знающие только Thread и synchronized, начинают изобретать очередь сами. В реальности этот паттерн — основа обработчиков событий, воркеров отправки писем, батчинга записи в БД; в проде чаще берут готовый ThreadPoolExecutor с его внутренней очередью. Бонусные темы: что происходит при отказе потребителя, ретраи, метрика глубины очереди как сигнал перегрузки.
Пример кода
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
Runnable POISON = () -> {};
// Producer
void produce(Runnable task) throws InterruptedException {
queue.put(task); // блокируется, если очередь полна
}
// Consumer (запустить в N потоках)
void consume() {
try {
while (true) {
Runnable task = queue.take(); // ждёт элемент
if (task == POISON) return; // graceful stop
task.run();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}Частые ошибки
- Берут неограниченную LinkedBlockingQueue и не могут объяснить, чем это грозит
- В ручной реализации проверяют условие через if вместо while, ловя spurious wakeup
- Глотают InterruptedException без восстановления флага прерывания