Буферизация данных в многопоточном приложении Java

Question

У меня есть многопоточное приложение, в котором есть один поток производителей и несколько потоков потребителей. Данные хранятся в общей потоковой безопасной коллекции и сбрасываются в базу данных, когда в буфере имеется достаточное количество данных.

Из Javadocs -

BlockingQueue<E> 

очередь, что дополнительно поддерживает операции, которые ожидают очереди, чтобы стать непустыми при получении элемента, и ждать пространства, чтобы стать доступными в очереди при хранении элемента ,

take() 

Извлекает и удаляет головку этой очереди, ожидая при необходимости, пока элемент не станет доступным.

Мои вопросы -

1. Есть еще одна коллекция, которая имеет метод E [принимать] (Int N)? то есть блокировка очереди ожидает, пока элемент не будет доступен. Я хочу , что он должен подождать, пока не будет доступно 100 или 200 элементов.
2. В качестве альтернативы, есть ли другой способ, который я мог бы использовать для решения проблемы без опроса?

Answer 1

Я думаю, что единственный способ либо продлить некоторую реализацию BlockingQueue, либо создать какой-либо полезный метод, используя take:

public <E> void take(BlockingQueue<E> queue, List<E> to, int max) 
     throws InterruptedException { 

    for (int i = 0; i < max; i++) 
     to.add(queue.take()); 
} 

Answer 2

Я не уверен, есть ли подобный класс в стандартной библиотеке, которая имеет take(int n) метод типа, но вы должны быть в состоянии обернуть по умолчанию BlockingQueue, чтобы добавить эту функцию, не слишком много хлопот, не вы думать?

Альтернативный сценарий должен был инициировать действие, в котором вы помещаете элементы в коллекцию, где установленный вами порог запускает промывку.

Answer 3

Метод drainTo не совсем то, что вы ищете, но будет ли он служить вашей цели?

http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/BlockingQueue.html#drainTo(java.util.Collection, целое)

EDIT

Вы могли бы реализовать несколько более производительную партия блокирует takemin используя комбинацию take и drainTo:

public <E> void drainTo(final BlockingQueue<E> queue, final List<E> list, final int min) throws InterruptedException 
{ 
    int drained = 0; 
    do 
    { 
    if (queue.size() > 0) 
     drained += queue.drainTo(list, min - drained); 
    else 
    { 
     list.add(queue.take()); 
     drained++; 
    } 
    } 
    while (drained < min); 
} 

Answer 4

Таким образом, это должна быть потоковая очередь, которая позволяет блокировать принятие произвольного количества элементов. Больше глаз, чтобы проверить код резьбы, будет правильным.

package mybq; 

import java.util.ArrayList; 
import java.util.LinkedList; 
import java.util.List; 

public class ChunkyBlockingQueue<T> { 
    protected final LinkedList<T> q = new LinkedList<T>(); 
    protected final Object lock = new Object(); 

    public void add(T t) { 
     synchronized (lock) { 
      q.add(t); 
      lock.notifyAll(); 
     } 
    } 

    public List<T> take(int numElements) { 
     synchronized (lock) { 
      while (q.size() < numElements) { 
       try { 
        lock.wait(); 
       } catch (InterruptedException e) { 
        Thread.currentThread().interrupt(); 
       } 
      } 
      ArrayList<T> l = new ArrayList<T>(numElements); 
      l.addAll(q.subList(0, numElements)); 
      q.subList(0, numElements).clear(); 
      return l; 
     } 
    } 
}