image
JAVA中的幾種主要的阻塞隊(duì)列
-
ArrayBlockingQueue: 基于數(shù)組實(shí)現(xiàn)的一個(gè)阻塞隊(duì)列榕订，在創(chuàng)建ArrayBlockingQueue對象必須指定容量大小渤滞。并且可以指定公平性和非公平性苏潜，默認(rèn)情況下為非公平性，即不保存等待時(shí)間最長的隊(duì)列有限能夠訪問隊(duì)列耕漱。 -
LinkedBlockingQueue: 基于鏈表實(shí)現(xiàn)的一個(gè)阻塞隊(duì)列龙优，在創(chuàng)建LinkedBlockingQueue對象如果不指定容量大小，則為Integer.MAX_VALUE. -
PriorityBlockingQueue: 以上2種隊(duì)列都是先進(jìn)先出隊(duì)列浇借，而PriorityBlockQueue卻不是捉撮，它會按照元素的優(yōu)先級對元素進(jìn)行排序，按照優(yōu)先級順序出隊(duì)妇垢，每次出隊(duì)的元素都是優(yōu)先級最高的元素巾遭。注意，此阻塞隊(duì)列為無界阻塞隊(duì)列闯估，即容量沒有上限灼舍。 -
DelayQueue: 基于PriorityBlockingQueue: 基于PriortyQueue, 一種延時(shí)阻塞隊(duì)列，DelayQueue中的元素只有當(dāng)其指定的延遲時(shí)間到了涨薪，才能夠從隊(duì)列中獲取到該元素骑素。DelayQueue也是一種無界隊(duì)列，因此往隊(duì)列中插入數(shù)據(jù)的操作（生產(chǎn)者）永遠(yuǎn)不會被阻塞刚夺，而只有獲取數(shù)據(jù)的操作（消費(fèi)者）才會被阻塞献丑。
阻塞隊(duì)列的幾個(gè)方法
- put(E e): 向隊(duì)尾存入元素，如果隊(duì)列滿了侠姑，則等待创橄。
- tack(): 從隊(duì)首取元素，如果隊(duì)伍為空莽红，則等待妥畏。
- offer(E e, long timeout, TimeUnil unit): 如果隊(duì)列滿了則等待一定時(shí)間，沒有插入成功則返回false,成功則返回true.
- poll(long timeout, TimeUnit unit): 如果隊(duì)列為空安吁，則等待一定時(shí)間醉蚁，當(dāng)時(shí)間期限達(dá)到時(shí)，如果未取到柳畔，則返回null.
實(shí)現(xiàn)
- 阻塞隊(duì)列實(shí)現(xiàn)
public class BlockingQueue {
private List<Object> queue = new LinkedList<>();
private int limit = 10;
public BlockingQueue(int limit){
this.limit = limit;
}
/**
* 當(dāng)隊(duì)列滿了的時(shí)候阻塞線程
*
* @param item
* @throws InterruptedException
*/
public synchronized void enqueue(Object item) throws InterruptedException {
while (this.queue.size() == this.limit) {
wait();
}
if (this.queue.size() < this.limit) {
notifyAll();
}
this.queue.add(item);
}
/**
* 當(dāng)隊(duì)列為空的時(shí)候阻塞線程
* @return
* @throws InterruptedException
*/
public synchronized Object dequeue() throws InterruptedException {
while (this.queue.size() == 0) {
wait();
}
if (this.queue.size() > 1) {
notifyAll();
}
return this.queue.remove(0);
}
}
- 使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者
public class ProducerConsumerPattern {
public static void main(String[] args) {
BlockingQueue shareQueue = new LinkedBlockingQueue<>();
Thread prodThread = new Thread(new Producer(shareQueue));
Thread consThread = new Thread(new Consumer(shareQueue));
prodThread.start();
consThread.start();
}
/**
* 生產(chǎn)者
*/
static class Producer implements Runnable {
private final BlockingQueue shareQueue;
public Producer(BlockingQueue shareQueue) {
this.shareQueue = shareQueue;
}
@Override
public void run() {
for (int i = 0; i < 10; i++){
System.out.println("Produced: " + i);
try {
shareQueue.put(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
/**
* 消費(fèi)者
*/
static class Consumer implements Runnable {
private final java.util.concurrent.BlockingQueue shareQueue;
public Consumer(BlockingQueue shareQueue) {
this.shareQueue = shareQueue;
}
@Override
public void run() {
while (true) {
try {
System.out.println("Consumer: " + shareQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
- 使用Lock和Condition實(shí)現(xiàn)阻塞隊(duì)列
public class BoundedBuffer {
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
final Object[] items = new Object[100];
int putptr, takeptr, count;
public void put(Object x) throws InterruptedException {
lock.lock();
try {
while (count == items.length)
notFull.await();
items[putptr] = x;
if (++putptr == items.length)
putptr = 0;
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0)
notEmpty.await();
Object x = items[takeptr];
if (++takeptr == items.length)
takeptr = 0;
--count;
notFull.signal();
return x;
} finally {
lock.unlock();
}
}
}
