java中的多个线程附加到队列

Question

我有多个线程正在运行，需要附加到同一个队列。这个队列被分成多个变量，所以有效地，我正在调用一个函数，该函数在某个位置 i 处附加到每个变量。从http://www.youtube.com/watch?v=8sgDgXUUJ68&feature=c4-overview-vl&list=PLBB24CFB073F1048E 显示，我为每种方法添加了锁，如此处所示。 http://www.caveofprogramming.com/java/java-multiple-locks/ 。为数组中的 100,000 个对象创建锁似乎并不有效。在java中，假设我有一个管理大量对象队列的对象。如何正确同步 addToQueue，而不会通过同步方法而牺牲性能，只是我要附加到的 floatQueue 和 intQueue 中的位置？

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

考虑使用 BlockingQueue (java.util.concurrent) http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html

Answer 2

java.util.concurrent.ConcurrentLinkedQueue 是一个非阻塞、无锁的队列实现。它使用比较和交换指令而不是锁定。

Answer 3

您最好使用Blocking Queue，而不是重新发明轮子。如果您这样做是为了练习，那么您可以使用两个锁，一个用于putting，一个用于getting。具体参考Blocking queue的源码。

附带说明一下，确保正确的并发性很棘手，因此不要在部署应用程序中使用队列，而应依赖 Java 的实现。

Answer 4

要保持并行数组“同步”，您需要同步。这是保持对数组及其关联索引的修改原子性的唯一方法。

您可能会考虑使用不可变元组对象而不是并行数组，为您的int 和float 提供final 字段，然后将它们添加到BlockingQueue 实现中，例如ArrayBlockingQueue。

如果有很多流失（频繁创建和丢弃对象），synchronized 可能会比元组执行得更好。您需要对它们进行概要分析才能确定。

Answer 5

我发现一个可能的解决方案是使用 AtomicIntegers，它类似于锁定，但运行的级别比同步的低得多。这是我的代码的更新副本，使用了 AtomicInteger。请注意，它仍然需要测试，但理论上应该足够了。

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;}
        }
    }
}

另外，作为参考，值得一读

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