CareerCup

Looked like a simplified version of java.util.concurrent.ArrayBlockingQueue which implements java.util.concurrent.BlockingQueue will suffice. Except there is a twist of multi-put that must be atomic.

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;

public class ArrayMultiPutBlockingBoundedQueue<E> implements MultiPutBlockingBoundedQueue<E> {

    public ArrayMultiPutBlockingBoundedQueue() {
        elements = null;
        count = 0;
    }

    @Override
    public synchronized void init(int capacity) throws Exception {
        if (isInitialized()) { // already initialized
            throw new IllegalStateException();
        }
        if (capacity <= 0) {
            throw new IllegalArgumentException();
        }

        elements = new Object[capacity];
        count = 0;
        start = 0;
        notifyAll();
    }

    @Override
    public synchronized E get() throws Exception {
        if (!isInitialized()) {
            throw new IllegalStateException();
        }

        while (isEmpty()) {
            try {
                wait();
            } catch (final InterruptedException e) {
            }
        }

        assert (!isEmpty());
        final E val = dequeue();
        notifyAll();
        return val;
    }

    @Override
    public synchronized void put(E obj) throws Exception {
        Objects.requireNonNull(obj);

        if (!isInitialized()) {
            throw new IllegalStateException();
        }

        while (isFull()) {
            try {
                wait();
            } catch (final InterruptedException e) {
            }
        }

        assert (!isFull());
        enqueue(obj);
        notifyAll();
    }

    @Override
    public synchronized void multiput(List<E> objs) throws Exception {
        if (!isInitialized()) {
            throw new IllegalStateException();
        }
        if (objs.size() > elements.length) {
            throw new IllegalArgumentException();
        }

        while (!hasCapacity(objs.size())) {
            try {
                wait();
            } catch (final InterruptedException e) {
            }
        }

        assert (hasCapacity(objs.size()));
        for (final E obj : objs) {
            Objects.requireNonNull(obj);
            enqueue(obj);
        }
        notifyAll();
    }

    private synchronized boolean isInitialized() {
        return (elements != null);
    }

    private synchronized boolean isEmpty() {
        return (count == 0);
    }

    private synchronized boolean isFull() {
        return (count == elements.length);
    }

    private synchronized boolean hasCapacity(int n) {
        return (count + n <= elements.length);
    }

    private synchronized int end() {
        return (start + count) % elements.length;
    }

    private synchronized void enqueue(E o) {
        final int idx = end();
        elements[idx] = o;
        // System.out.format("enqueue msg\"%s\" at index \"%d\"%n",
        // o.toString(), idx); // DEBUG
        count++;
    }

    private synchronized E dequeue() {
        @SuppressWarnings("unchecked")
        final E val = (E) elements[start];
        elements[start] = null;
        // System.out.format("dequeue msg\"%s\" at index \"%d\"%n",
        // val.toString(), start); // DEBUG
        start = (start + 1) % elements.length;
        count--;
        return val;
    }

    private Object[] elements;
    private int count;
    private int start;
}

Here is an ad-hoc test driver to test (note that driver will hang at the end, because consumer threads will be blocked for producers to insert more elements, but producer threads would have ended):

public class TestDriver {

    private static class Producer implements Runnable {
        public Producer(ArrayMultiPutBlockingBoundedQueue<String> queue, int id) {
            this.queue = queue;
            this.id = id;
        }

        @Override
        public void run() {
            try {
                while (counter < COUNTER_MAX) {
                    if (ThreadLocalRandom.current().nextBoolean()) {
                        counter++;
                        final String msg = "Producer[" + id + "] : Message #: " + counter;
                        System.out.format("inserting: %s %n", msg); // DEBUG
                        queue.put(msg);
                    } else {
                        final int count = ThreadLocalRandom.current().nextInt(1, 4);
                        final List<String> msgs = new ArrayList<String>(count);
                        final String msgPrefix = "Producer[" + id + "] : Message #: ";
                        for (int i = 0; i < count; i++) {
                            counter++;
                            msgs.add(msgPrefix + counter);
                        }
                        System.out.format("inserting: %sfrom %d to %d %n", msgPrefix, counter - count + 1, counter); // DEBUG
                        queue.multiput(msgs);
                    }
                }
            } catch (final Exception e) {
                e.printStackTrace();
            }
        }

        private final ArrayMultiPutBlockingBoundedQueue<String> queue;
        private final int id;
        private int counter = 0;
        private static final int COUNTER_MAX = 32;
    }

    private static class Consumer implements Runnable {
        public Consumer(ArrayMultiPutBlockingBoundedQueue<String> queue, int id) {
            this.queue = queue;
            this.id = id;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    final String msg = queue.get();
                    System.out.format("Consumer[%d] extracted msg: %s %n", id, msg); // DEBUG
                }
            } catch (final Exception e) {
                e.printStackTrace();
            }
        }

        private final ArrayMultiPutBlockingBoundedQueue<String> queue;
        private final int id;
    }

    public static void main(String[] args) throws Exception {
        int num = 0;
        final int MIN_THREADS = 3;
        final int MAX_THREAD = 8;
        final int QUEUE_CAPACITY = 32;

        if (args.length >= 0) {
            try {
                num = Integer.parseInt(args[0]);
            } catch (final Exception e) {
            }
        }
        if (num < MIN_THREADS)
            num = MIN_THREADS;
        if (num > MAX_THREAD)
            num = MAX_THREAD;

        final ArrayMultiPutBlockingBoundedQueue<String> queue = new ArrayMultiPutBlockingBoundedQueue<String>();
        queue.init(QUEUE_CAPACITY);
        Thread.sleep(1000);

        for (int i = 0; i < num; i++) {
            (new Thread(new Producer(queue, i))).start();
            (new Thread(new Consumer(queue, i))).start();
        }
    }

}

class MultiPutBlockingBoundedQueueImpl<E> implements MultiPutBlockingBoundedQueue<E> {
    private E[] buffer;
    private int position;

    @Override
    public void init(int capacity) throws Exception {
        if (buffer != null) {
            throw new IllegalStateException("Already initialized");
        } else if (capacity <= 0) {
            throw new IllegalArgumentException("Capacity must be positive number");
        } else {
            buffer = (E[]) new Object[capacity];
        }
    }

    @Override
    public E get() throws Exception {
        if (buffer == null) {
            throw new IllegalStateException("Not initialized");
        }
        synchronized (buffer) {
            while (position == 0) {
                buffer.wait();
            }
            position--;
            return buffer[position + 1];
        }

    }

    @Override
    public void put(E obj) throws Exception {
        multiput(new ArrayList<E>(){{add(obj);}});
    }

    @Override
    public void multiput(List<E> objs) throws Exception {
        if (buffer == null) {
            throw new IllegalStateException("Not initialized");
        }

        synchronized (buffer) {
            while (position >= buffer.length - 1 - objs.size()) {
                buffer.wait();
            }

            for (E obj : objs) {
                position++;
                buffer[position] = obj;
            }
            buffer.notifyAll();
        }
    }
}

class MultiPutBlockingBoundedQueueImpl<E> implements MultiPutBlockingBoundedQueue<E> {
    private E[] buffer;
    private int position;

    @Override
    public void init(int capacity) throws Exception {
        if (buffer != null) {
            throw new IllegalStateException("Already initialized");
        } else if (capacity <= 0) {
            throw new IllegalArgumentException("Capacity must be positive number");
        } else {
            buffer = (E[]) new Object[capacity];
        }
    }

    @Override
    public E get() throws Exception {
        if (buffer == null) {
            throw new IllegalStateException("Not initialized");
        }
        synchronized (buffer) {
            while (position == 0) {
                buffer.wait();
            }
            position--;
            return buffer[position + 1];
        }

    }

    @Override
    public void put(E obj) throws Exception {
        multiput(new ArrayList<E>(){{add(obj);}});
    }

    @Override
    public void multiput(List<E> objs) throws Exception {
        if (buffer == null) {
            throw new IllegalStateException("Not initialized");
        }

        synchronized (buffer) {
            while (position >= buffer.length - 1 - objs.size()) {
                buffer.wait();
            }

            for (E obj : objs) {
                position++;
                buffer[position] = obj;
            }
            buffer.notifyAll();
        }
    }
}

public class MultiPutBoundedQueue {
    private LinkedList list;
    private int maxCapacity;
    private final Lock lock = new ReentrantLock();
    private final Condition readCondition = lock.newCondition();
    private final Condition writeCondition = lock.newCondition();
    private final Condition writeCondition2 = lock.newCondition();
    private volatile int countDownLatch;
    private volatile boolean spaceAvailable = true;


    public void init(int capacity) {
        assertCapacity(capacity);

        list = new LinkedList();
        maxCapacity = capacity;
    }

    private void assertCapacity(int capacity) {
        if (capacity <= 0)
            throw new IllegalArgumentException("Capacity must be positive number");
    }


    public Object get() throws InterruptedException {
        assertInitialized();

        lock.lock();
        try {
            while (list.size() == 0) {
                readCondition.await();
            }
            Object o = list.remove();
            if (countDownLatch > 0) {
                countDownLatch--;
            }
            System.out.println("[-1] Read 1, " + list.size());
            if (countDownLatch == 0) {
                writeCondition2.signalAll();
            }
            return o;
        } finally {
            lock.unlock();
        }
    }


    public void put(Object obj) throws InterruptedException {
        multiput(Arrays.asList(obj));
    }


    public void multiput(List objs) throws InterruptedException {
        assertInitialized();
        assertOverCapacity(objs); // if objs.size() > maxCapacity

        lock.lock();
        try {
            while (!spaceAvailable) {
                writeCondition.await();
            }

            if (list.size() + objs.size() > maxCapacity) {
                countDownLatch = list.size() + objs.size() - maxCapacity;
                while (countDownLatch > 0) {
                    spaceAvailable = false;
                    writeCondition2.await();
                }
                spaceAvailable = true;
            }
            list.addAll(objs);


            System.out.println(String.format("[%s] Added %s, ", Thread.currentThread().getName(), objs.size()) + list.size());
            writeCondition.signalAll();
            readCondition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    private void assertOverCapacity(List objs) {
        if (objs.size() > maxCapacity) {
            throw new IllegalArgumentException("Max number of items is: " + maxCapacity);
        }
    }


    private void assertInitialized() {
        if (list == null) {
            throw new IllegalStateException("MultiPutBlockingBoundedQueue has not been initialized with .init(...)");
        }
    }


    public static void main(String[] args) throws Exception {
        final MultiPutBoundedQueue multiPutBoundedQueue = new MultiPutBoundedQueue();
        multiPutBoundedQueue.init(10);
        Thread put = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    int counter = 0;
                    while (true) {
                        Thread.currentThread().setName("+1/" + counter++);
                        multiPutBoundedQueue.put("");
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread mput = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    int counter = 0;
                    while (true) {
                        Thread.currentThread().setName("+5/" + counter++);
                        multiPutBoundedQueue.multiput(Arrays.asList(1, 2, 3, 4, 5));
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread read = new Thread(new Runnable() {
            @Override
            public void run() {
                Thread.currentThread().setName("-1");
                try {
                    while (true) {
                        multiPutBoundedQueue.get();
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        put.start();
        mput.start();
        read.start();
    }

}