Как избежать впустую расходования ресурсов при чтении из BlockingQueue Java?

Question

[Dik_Nik]

Добрый день!
Имеется следующая архитектура сервиса:

Как видно из рисунка, имеется несколько тредов-продюсеров, которые пишут каждый в свою очередь, и один тред-консьюмер, который читает из всех очередей.
Проблема в том, что консьюмер работает в бесконечном цикле и впустую расходует большое количество процессорного времени и значительное количество памяти, на создание итератора.
Может кто нибудь знает, есть ли возможность засыпать, если очереди пустые и просыпаться, если в очередях что-то появилось, например как это сделано в nio библиотеки с Селектором?

Comments

[mayton2019]

Для потоков есть механизмы wait и notify. Причем давно.

[Dik_Nik]

mayton2019, Хотелось бы без захвата монитора

[Dmitry Roo]

Почему бы вам не добавить код? Хотя бы demo / proof of concept

[Dik_Nik]

Dmitry Roo, проблемную часть кода приложил на рисунке
Какой код еще тут будет полезный?

[Dmitry Roo]

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

[сергей кузьмин]

Dik_Nik, вам надо добавить любой pubsub (встроенные можно найти на гитхабе хотя чаще используют внешние напр. gcp и сделать push subscriber (который передает callback который будет вызван ) и publisher

Answer 1

[Dmitry Roo]

Кажется, вашу проблему полностью решает project reactor

Вот небольшое демо:

@Test
    void testMultiproducer() throws InterruptedException {

        Sinks.Many<Integer> sinksA = Sinks.many().multicast().onBackpressureBuffer();
        Sinks.Many<Integer> sinksB = Sinks.many().multicast().onBackpressureBuffer();
        Sinks.Many<Integer> sinksC = Sinks.many().multicast().onBackpressureBuffer();

        Sinks.Many<Integer> sinkCommon = Sinks.many()
                .multicast()
                .onBackpressureBuffer();

        sinkCommon.asFlux()
                .mergeWith(sinksA.asFlux().delayElements(Duration.ofMillis(100)))
                .mergeWith(sinksB.asFlux().delayElements(Duration.ofMillis(200)))
                .mergeWith(sinksC.asFlux().delayElements(Duration.ofMillis(300)))
                .subscribe(e -> log.info("Element {}", e));  // Your consumer
        
        IntStream.range(0, 100)
                .forEach(sinksA::tryEmitNext);  // Your producers

        IntStream.range(1000, 1100)
                .forEach(sinksB::tryEmitNext);

        IntStream.range(2000, 2100)
                .forEach(sinksC::tryEmitNext);

        Thread.sleep(10000); // Is necessary for test only
    }