Multiple threads in java appending to a queue

Question

I have multiple threads running which need to append to the same queue. This queue is split up into multiple variables, so effectively, I am calling a function, which appends to each variable at some position i. From http://www.youtube.com/watch?v=8sgDgXUUJ68&feature=c4-overview-vl&list=PLBB24CFB073F1048E I was shown to add locks for each method, as shown here. http://www.caveofprogramming.com/java/java-multiple-locks/ . It does not seem effective to create a lock for 100,000 objects in an array. In java, let's say that I have an object which manages a large queue of objects. How can I properly synchronize addToQueue, without sacrificing performance by synchronizing the method, just the position in the floatQueue and intQueue that I am appending to?

class QueueManager {
    private float[] floatQueue;
    private int[] intQueue;

    private int currentQueueSize;
    private int maxQueueSize;

    QueueManager(int sizeOfQueue) {
        intQueue = new int[sizeOfQueue];
        floatQueue = new float[sizeOfQueue];

        currentQueueSize = 0;
        maxQueueSize = sizeOfQueue;
    }

    public boolean addToQueue(int a, float b) {
        if (currentQueueSize == maxQueueSize) {
            return false;
        }
        else{
            intQueue[currentQueueSize] = a;
            floatQueue[currentQueueSize] = b;

            currentQueueSize++;

            return true;
        }
    }
}

Answer 1

java.util.concurrent.ConcurrentLinkedQueue is a non-blocking, lock-free queue implementation. It uses compare-and-swap instructions instead of locking.

Answer 2

Consider using BlockingQueue (java.util.concurrent) http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html

Answer 3

You are better off using Blocking Queue than reinventing the wheel. In case you are doing this for practice then you can use two locks one for putting and one of getting. Refer to the source of Blocking queue for that.

Just a side note getting concurrency right is tricky so do not use your queue in deployment application and rather rely on implementation of Java.

Answer 4

To keep the parallel arrays "in synch", you need synchronization. It's the only way to preserve the atomicity of modifications to both arrays and their associated index.

Rather than parallel arrays, you might consider an immutable tuple object, with final fields for your int and float, then adding those to a BlockingQueue implementation, like ArrayBlockingQueue.

If there's a lot of churn (frequent creation and discarding of objects), synchronized will probably perform better than tuples. You'd need to profile them to be sure.

Answer 5

One possible solution I found would be to use AtomicIntegers, which is similar to locking, but runs on a much lower level than what synchronized does. Here is an updated copy of my code, with an AtomicInteger used. Please note that it still needs to be tested, but in theory, it should be sufficient.

import java.util.concurrent.atomic.AtomicInteger;

class ThreadSafeQueueManager {
    private float[] floatQueue;
    private int[] intQueue;

    private AtomicInteger currentQueueSize;
    private int maxQueueSize;

    QueueManager(int sizeOfQueue) {
        intQueue = new int[sizeOfQueue];
        floatQueue = new float[sizeOfQueue];

        currentQueueSize = new AtomicInteger(0);
        maxQueueSize = sizeOfQueue;
    }

    public boolean addToQueue(int a, float b) {
        if (currentQueueSize.get() == maxQueueSize) {
            return false;
        }
        else{
            try {
                // Subtract one so that the 0th position can be used
                int position = currentQueueSize.incrementAndGet() - 1;
                intQueue[position] = a;
                floatQueue[positions] = b;

                return true;
            }
            catch (ArrayIndexOutOfBoundsException e) { return false;}
        }
    }
}

Also, for reference, it is worth reading

https://www.ibm.com/developerworks/java/library/j-jtp11234/
http://www.ibm.com/developerworks/library/j-jtp04186/