按 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

从输出中，您可以清楚地看到第一个元素被删除，而最近的元素被添加到队列中。

我的疑问：这是好的解决方案吗？还是有其他更好的解决方案，比这更好？

编辑：在这种频繁删除的场景中，ArrayBlockingQueue 是否比 LinkedBlockingQueue 好？

Answer 1

让我们避免重新发明轮子，只需使用具有固定容量的LinkedBlockingQueue，它是线程安全的FIFOBlockingQueue。更多详情here.

您的代码的问题在于您执行以下操作不是原子的，因此您可能会面临竞争条件问题：

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
________

Answer 2

我必须首先承认我从未使用过并发包中的 BlockingQueue，但我之前做过多线程编程。

我认为这里有问题：

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

如果有多个线程同时运行此方法，则多个线程可以执行此检查，并且在它们采取行动之前，它可以评估大量线程为真。因此，在这种情况下，remove() 的调用次数可能比您预期的要多。

基本上，如果您想保持这种逻辑，则必须找到一种方法来确保在检查大小然后执行之间另一个线程不可能更改队列的大小对其进行操作，例如删除元素。

解决此问题的一种方法可能是将其包装在同步块中，如下所示：

synchornized (queue) {
  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("________");
}

这确保了queue 在您检查其当前大小后不会被其他线程访问。请记住，您同步的内容越多，其他线程在对其执行操作之前必须等待的时间越长，这可能会减慢您的程序。您可以阅读有关此关键字的更多信息here。