BoundedPriorityBlockingQueue - threadsicher, blockierend und begrenzt?

Question

Gibt es in Java eine Priority-Queue, die wie LinkedBlockingQueue funktioniert?

PriorityBlockingQueue blockiert nicht, da es unbegrenzt ist.

Answer 1

können Sie versuchen, MinMaxPriorityQueue von Google Guava und legen Sie eine maximale Größe als nächste:

Queue<User> users = Queues.synchronizedQueue(
    MinMaxPriorityQueue.orderedBy(userComparator) 
     .maximumSize(1000) 
     .create() 
); 

NB: Als MinMaxPriorityQueue ist nicht Thread-sicher Sie den Dekorateur Queues.synchronizedQueue(Queue) verwenden müssen, so dass es machen fadensicher.

Da Sie eine BlockingQueue benötigen, müssen Sie den Dekorator selbst implementieren, was nicht schwer zu implementieren ist.

Hier ist, wie es aussehen sollte:

public class SynchronizedBlockingQueue implements BlockingQueue { 

    private final BlockingQueue queue; 

    public SynchronizedBlockingQueue(BlockingQueue queue) { 
     this.queue = queue; 
    } 

    @Override 
    public synchronized boolean add(final Object o) { 
     return this.queue.add(o); 
    } 

    @Override 
    public synchronized boolean offer(final Object o) { 
     return this.offer(o); 
    } 
    ... 
} 

Dann wird der Code in Ihre erstellen BlockingQueue wird:

BlockingQueue<User> users = new SynchronizedBlockingQueue(
    MinMaxPriorityQueue.orderedBy(userComparator) 
     .maximumSize(1000) 
     .create() 
); 

Answer 2

Wenn Sie nicht brauchen, eine ausgewachsene BlockingQueue Interface-Implementierung dann Sie können Semaphore und etwas ähnliches (in Kotlin) verwenden:

interface BlockingBag<E: Any> { 
    @Throws(InterruptedException::class) 
    fun put(element: E) 
    @Throws(InterruptedException::class) 
    fun take(): E 
} 
class BlockingPriorityBag<E: Any>(val capacity: Int) : BlockingBag<E> { 
    init { 
     require(capacity >= 1) { "$capacity must be 1 or greater" } 
    } 
    private val queue = PriorityBlockingQueue<E>() 
    private val semaphore = Semaphore(capacity) 
    override fun take(): E { 
     val item = queue.take() 
     semaphore.release() 
     return item 
    } 
    override fun put(element: E) { 
     semaphore.acquire() 
     queue.put(element) 
    } 
}