The scenario is about processing 'Message' objects. Producer creates them and Consumer does the consumption. class Message{...}class MessageBatch{    public MessageBatch(Collection<Message> messages){...}    public List<Message> getMessages(){...};    //more utility methods like getMessageIds(), etc.}The following is a pipeline object (queue) that connects multiple producer threads and multiple consumer threads. Both my producers & consumers are I/O bound (database) and that justifies this attempt.API requirements[A] Ability to add & get batch of Message objects.[B] Consumer has a way to realize that no more elements are expected.public class MessagePipeline {    //[1]; to support API requirement [B]    public static final MessageBatch SENTINEL_BATCH = new MessageBatch(Collections.<Message>emptyList());    private BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();    // to support API requirement [B]    private CountDownLatch producersWorkCompletionLatch;    public MessagePipeline(int numOfProducerThreads) {        this.producersWorkCompletionLatch = new CountDownLatch(numOfProducerThreads);    }    public boolean add(MessageBatch batch) {        isWriteAllowed(); // UPDATE: this step is misleading and thinking of removing this.        return messageQueue.addAll(batch.getMessages());    }    // returns a batch containing up to maxSize elements.     public MessageBatch takeBatch(int maxSize) {        List<Message> messages = drainIfMoreElementsExpected(new ArrayList<Message>(), maxSize);        return messages.isEmpty() ? SENTINEL_BATCH : new MessageBatch(messages);    }    //[2]: Producer should invoke this at the end.    public void notifyCompletion() {        producersWorkCompletionLatch.countDown();    }    public boolean isMoreElementsExpected() {        boolean isExpected = !(producersWorkCompletionLatch.getCount() == 0 && messageQueue.size() == 0);        return isExpected;    }    //[3] : This is CPU bound waiting. TODO: Make it I/O bound.    private List<Message> drainIfMoreElementsExpected(List<Message> messages, int maxSize) {        while (messages.isEmpty() && isMoreElementsExpected()) {            messageQueue.drainTo(messages, maxSize);        }        return messages;    }    /**     * Assures write-operations 'happen-before' producer-threads-completion.     */    private void isWriteAllowed(){        if(producersWorkCompletionLatch.getCount() == 0){            throw new UnsupportedOperationException(New elements are not accepted after all producer threads notify of completion);        }    }}Questions I'm looking forward to code review in general.Do approaches [1] and [2] seem right?[3]: Am I right that it is CPU bound? How do I make it I/O bound waiting? Introducing locks seem to make it more complicated. I'm in the middle of writing unit tests. It's good to know if I missed anything because unit-testing multi-threaded code tells you that the code is broken only one in million times and doesn't tell you why!
Queue that connects multiple producers and multiple consumers
java;multithreading;concurrency