public boolean run ( ) { ExecutorService executors = Executors . newFixedThreadPool ( producers . size ( ) + consumers . size ( ) ) ; runningProducers . set ( producers . size ( ) ) ; //create the producers for ( Producer < V > producer : producers ) { producer . setOwner ( this ) ; executors . submit ( new ProducerThread ( producer ) ) ; } //create the consumers for ( java . util . function . Consumer < ? super V > consumer : consumers ) { executors . submit ( new ConsumerThread ( consumer ) ) ; } //give it some more time to process while ( runningProducers . get ( ) > 0 || ! queue . isEmpty ( ) ) { Threads . sleep ( 10 ) ; } executors . shutdown ( ) ; try { executors . awaitTermination ( Integer . MAX_VALUE , TimeUnit . SECONDS ) ; } catch ( InterruptedException e ) { logWarn ( e ) ; return false ; } return true ; }
Starts execution of the broker .