按先进先出顺序固定容量的线程安全收集

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

你需要将它包装成一个​​块或使用一个明确的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

我必须首先承认，我从来没有从concurrent包使用的BlockingQueue的，但我之前完成了多线程编程。

我认为这里有一个问题：

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

如果有同时运行这种方法多线程，则多线程可以做这样的检查，才送它可以评估为true，相当数量人行动。所以在这种情况下，remove()可能比您预期的要多。

基本上，如果你想保持你的逻辑，你将不得不找到一种方法来确保另一个线程不可能在检查大小的时候改变队列的大小，然后对其执行操作，如删除元素。解决这个问题

一种方式是将其包装在synchronized块，像这样：

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。