BoundedPriorityBlockingQueue - 線程安全，阻塞和有界？

Question

是否有Java中的優先級隊列，其行爲類似於LinkedBlockingQueue？

PriorityBlockingQueue沒有被阻塞，因爲它是無界的。

Answer 1

您可以從谷歌番石榴嘗試MinMaxPriorityQueue並設置爲下一個最大尺寸：

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

注意：作爲MinMaxPriorityQueue不是線程安全的，你需要使用裝飾Queues.synchronizedQueue(Queue)允許使它線程安全。

，因爲你需要一個BlockingQueue你將必須實現自己的裝飾這是不難實現的。

這是應該的樣子：

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

然後代碼來創建BlockingQueue將是：

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

Answer 2

如果您不需要一個完全成熟的BlockingQueue接口實現再你可以使用Semaphore和類似的東西（在Kotlin中）：

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