以FIFO顺序固定容量的线程安全收集

Question

问题:维护一个具有固定容量的集合(比如2个元素),可以同时访问100多个线程.

始终存储最近线程中的最新元素.存储后,编写一个方法来检查所有这些元素是否重复.

我的解决方案:BlockingQueue具有固定容量并实现自定义添加方法.

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.Iterator;

public class FixedBlockingQueue<T extends Object> {

    final BlockingQueue<T> queue;
    private int capacity;

    public FixedBlockingQueue(int capacity){
        super();
        this.capacity = capacity;
        queue = new ArrayBlockingQueue<T>(capacity);
        System.out.println("Capactiy:"+this.capacity);
    }
    public void addElement(T element){
        try{
            if ( queue.size() > capacity - 1 ){
                queue.remove();         
            }
            queue.put(element);
            for ( Iterator it = queue.iterator(); it.hasNext();){
                System.out.println(it.next());
            }
            System.out.println("________");
        }catch(Exception err){
            err.printStackTrace();
        }
    }

    public static void main(String args[]){
        FixedBlockingQueue<Integer> f = new FixedBlockingQueue<Integer>(2);
        for ( int i=0; i< 10; i++){
            f.addElement(i);
        }

    }   
}

输出:

0
________
0
1
________
1
2
________
2
3
________
3
4
________
4
5
________
5
6
________
6
7
________
7
8
________
8
9

从输出中,您可以清楚地看到第一个元素被删除,并且最近的元素被添加到Queue中.

我的疑问:这是一个很好的解决方案吗？还是有其他好的解决方案,比这更好吗？

编辑:在这种频繁删除的情况下,是否ArrayBlockingQueue优于LinkedBlockingQueue？

Answer 1

让我们避免重新发明轮子,只需使用LinkedBlockingQueue固定容量,它是一个线程安全FIFO BlockingQueue.更多细节在这里.

您的代码的问题在于您不是以原子方式执行以下操作,以便您可以面对竞争条件问题:

if ( queue.size() > capacity - 1 ){
    queue.remove();         
}
queue.put(element);

您需要将其包装到synchronized块中或使用显式Lock来保护它,因为它是一个关键部分,我们不希望多个线程同时调用它.

以下是如何通过以下方式完成BlockingQueue:

BlockingQueue queue = new LinkedBlockingQueue(2);
for ( int i=0; i< 10; i++){
    // Try to add the object and return immediately if it is full
    // then if it could not be added,
    // remove the last element of the queue and try again
    while (!queue.offer(i, 0L, TimeUnit.MICROSECONDS)) {
        queue.remove();
    }
    for ( Iterator it = queue.iterator(); it.hasNext();){
        System.out.println(it.next());
    }
    System.out.println("________");
}

输出:

0
________
0
1
________
1
2
________
2
3
________
3
4
________
4
5
________
5
6
________
6
7
________
7
8
________
8
9
________