Потребитель-производитель с высоким разрешением

Question

Я хотел бы знать, какой был бы лучший механизм для реализации нескольких сценариев Producer - single Consumer, где мне нужно постоянно обновлять текущее количество необработанных запросов.

Моя первая мысль была использовать ConcurrentLinkedQueue:

public class SomeQueueAbstraction { 

    private Queue<SomeObject> concurrentQueue = new ConcurrentLinkedQueue<>(); 
    private int size; 

    public void add(Object request) { 
     SomeObject object = convertIncomingRequest(request); 
     concurrentQueue.add(object); 
     size++; 

    } 

    public SomeObject getHead() { 
     SomeObject object = concurrentQueue.poll(); 
     size--; 
    } 

    // other methods 

Проблема с этим состоит в том, что я должен явно синхронизировать add и size ++, а также на poll и size--, чтобы всегда иметь точные size, что делает ConccurentLinkedQueue бессмысленно для начала.

Каким будет наилучший способ достижения максимально возможной производительности при сохранении согласованности данных?

Вместо этого я использую ArrayDequeue и явно синхронизирую, или есть лучший способ достичь этого?

Существует своего родом подобный вопрос/ответ здесь:

java.util.ConcurrentLinkedQueue

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

Примечание: Я вычисляю размер явно, потому что временная сложность для встроенного метода .size() - это O (n).

Note2: Я также обеспокоен тем, что метод getSize(), который я явно не написал, добавит еще больше конфликтов. Его можно было бы назвать относительно часто.

Я ищу наиболее эффективный способ обработки нескольких производителей - один потребитель с частыми вызовами getSize().

Альтернативное предложение: Если в структуре SomeObject был элемент, то я мог бы получить текущий размер из ConcurrentLinkedQueue.poll(), и только блокирование должно выполняться внутри механизма для создания такого идентификатора. Теперь добавить и получить можно правильно использовать без дополнительной блокировки. Как это будет в качестве альтернативы?

Answer 1

Таким образом, необходимо сообщить текущее количество необработанных запросов. И это часто запрашивается, что действительно делает ConcurrentLinkedQueue.size() непригодным.

Это можно сделать с помощью AtomicInteger: он быстрый и всегда как можно ближе к текущему количеству необработанных запросов.

Вот пример, отметят некоторые небольшие обновления для того, чтобы зарегистрированного размера является точным:

import java.util.Queue; 
import java.util.concurrent.ConcurrentLinkedQueue; 
import java.util.concurrent.atomic.AtomicInteger; 

public class SomeQueueAbstraction { 

    private final Queue<SomeObject> concurrentQueue = new ConcurrentLinkedQueue<>(); 
    private final AtomicInteger size = new AtomicInteger(); 

    public boolean add(Object request) { 

     SomeObject object = convertIncomingRequest(request); 
     if (concurrentQueue.add(object)) { 
      size.incrementAndGet(); 
      return true; 
     } 
     return false; 
    } 

    public SomeObject remove() { 

     SomeObject object = concurrentQueue.poll(); 
     if (object != null) { 
      size.decrementAndGet(); 
     } 
     return object; 
    } 

    public int getSize() { return size.get(); } 

    private SomeObject convertIncomingRequest(Object request) { 
     return new SomeObject(getSize()); 
    } 

    class SomeObject { 
     int id; 
     SomeObject(int id) { this.id = id; } 
    } 
} 

Answer 2

Вы можете использовать явную блокировку, что означает, что вам, вероятно, не понадобится параллельная очередь.

public class SomeQueueAbstraction { 

    private Queue<SomeObject> queue = new LinkedList<>(); 
    private volatile int size; 
    private Object lock = new Object(); 

    public void add(Object request) { 
     SomeObject object = convertIncomingRequest(request); 
     synchronized(lock) { 
      queue.add(object); 
      size++; 
     } 
    } 

    public SomeObject getHead() { 
     SomeObject object = null; 
     synchronized(lock) { 
      object = queue.poll(); 
      size--; 
     } 
     return object; 
    } 

    public int getSize() { 
     synchronized(lock) { 
      return size; 
     } 
    } 

    // other methods 
} 

Таким образом, добавление/удаление элементов в/из очереди и обновление size будет сделано безопасно.