哪个是java中更有效的线程安全队列

Question

有一个简单的任务：许多线程调用MyClass.add() 函数，一个线程尝试为它们服务。
我的问题：哪个是更好或更有效的解决方案？

第一种方法：CopyOnWriteArrayList

@Singleton
public class myClass {

    List<myType> list = new CopyOnWriteArrayList<myType>();
    boolean isRunning = false;

    //this is called from many threads
    public void add(myType x){
        list.add(x);
    }


    //this is called from 1 thread
    public void start(){
        if (isRunning) return;
        isRunning = true;

        while (!list.isEmpty()) {
            myType curr = list.remove(0);

            //do something with curr...
        }
        isRunning = false;
    }
}

使用简单锁的第二种方法：

@Singleton
public class myClass {

    List<myType> list = new ArrayList<myType>();
    boolean isRunning = false;
    private final Lock _mutex = new ReentrantLock(true);

    //this is called from many threads
    public void add(myType x){
        _mutex.lock();
        list.add(x);
        _mutex.unlock();
    }


    //this is called from 1 thread
    public void start(){
        if (isRunning) return;
        isRunning = true;

        while (!list.isEmpty()) {
            _mutex.lock();
            myType curr = list.remove(0);
            _mutex.unlock();

            //do something with curr...
        }
        isRunning = false;
    }
}

第三种方法：ConcurrentLinkedQueue

@Singleton
public class myClass {

    ConcurrentLinkedQueue<myType> list = new ConcurrentLinkedQueue<myType>();
    boolean isRunning = false;

    //this is called from many threads
    public void add(myType x){
        list.add(x);
    }


    //this is called from 1 thread
    public void start(){
        if (isRunning) return;
        isRunning = true;

        while (!list.isEmpty()) {
            //list cannot be empty at this point: other threads can't remove any items 
            myType curr = list.poll();

            //do something with curr...
        }
        isRunning = false;
    }
}

这是最初的错误解决方案。我不知道为什么它有时会给出（>100 个线程）ConcurrentModificationException（尽管有迭代器和“同步”）：

@Singleton
public class myClass {

    List<myType> list = Collections.synchronizedList(new ArrayList<myType>());
    boolean isRunning = false;

    //this is called from many threads
    public void add(myType x){
        synchronized(list) {
            list.add(x);
        }
    }


    //this is called from 1 thread
    public void start(){
        if (isRunning) return;
        isRunning = true;

        for (ListIterator<myType> iter = list.listIterator(); iter.hasNext();){
            myType curr = iter.next();

            //do something with curr...

            synchronized(list) {
                iter.remove(); //sometime it gives ConcurrentModificationException!
            }
        }
        isRunning = false;
    }
}

Answer 1

一般规则是：最适合您的问题。

锁变体使一切都变慢了很多，因为如果所有线程进入锁部分，它们就会被搁置，即使不需要它（如果有 5 个元素，5 个线程可以同时轮询它们，只有第6个必须等待）。但是，如果您有一个永远无法共享的单一资源（例如网络连接或文件），则此解决方案很好。

如果您的线程很少写入但经常读取，则 CopyOnWriteArrayList 是最佳解决方案。与写入相关的成本要高得多，这可以通过更快的读取来补偿（与 ConcurrentLinkedQueue 相比）。但是你的代码主要是写的，所以这对你来说不是一个好的解决方案。

如果读写量大致相等，ConcurrentLinkedQueue 是最好的解决方案，因此得名 Queue。所以它应该最适合你的情况。

此外，您的代码中存在严重错误：

while (!list.isEmpty()) {
  myType curr = list.poll();

该列表只是保证每次调用都是原子完成的，但您的代码不会因为您使用它而自动线程安全。在此示例中，列表可能已经在 isEmpty() 和 poll() 之间进行了修改，因此它在 isEmpty() 调用中可能有 1 个元素，但在您轮询后就没有了。 ConcurrentLinkedQueue 通过返回 null 来优雅地处理此问题，而不是由您的代码处理。所以正确的形式是：

myType curr;
while ((curr = list.poll()) != null) {

由于轮询是原子调用，因此是线程安全的，它要么返回一个元素，要么不返回。由于线程，之前发生的事情和之后发生的事情是不确定的，但您可以确定这个单一调用（在后台执行更多操作）将始终完美运行。

remove(0) 调用也是如此，如果最后一个元素已被另一个线程删除，它可能会抛出 IndexOutOfBoundsException。