快捷導(dǎo)航

java并發(fā)學(xué)習(xí)之BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者詳解

更新時(shí)間：2017年11月07日 17:36:50 作者：toMatser

這篇文章主要介紹了java并發(fā)學(xué)習(xí)之BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者詳解，具有一定參考價(jià)值，需要的朋友可以了解下。

1.介紹

阻塞隊(duì)列 (BlockingQueue)是Java util.concurrent包下重要的數(shù)據(jù)結(jié)構(gòu)，BlockingQueue提供了線程安全的隊(duì)列訪問(wèn)方式：當(dāng)阻塞隊(duì)列進(jìn)行插入數(shù)據(jù)時(shí)，如果隊(duì)列已滿，線程將會(huì)阻塞等待直到隊(duì)列非滿；從阻塞隊(duì)列取數(shù)據(jù)時(shí)，如果隊(duì)列已空，線程將會(huì)阻塞等待直到隊(duì)列非空。并發(fā)包下很多高級(jí)同步類的實(shí)現(xiàn)都是基于BlockingQueue實(shí)現(xiàn)的。

JDK7提供了以下7個(gè)阻塞隊(duì)列：

ArrayBlockingQueue ：由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。
LinkedBlockingQueue ：由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列。
PriorityBlockingQueue ：支持優(yōu)先級(jí)排序的無(wú)界阻塞隊(duì)列。
DelayQueue：使用優(yōu)先級(jí)隊(duì)列實(shí)現(xiàn)的無(wú)界阻塞隊(duì)列。
SynchronousQueue：不存儲(chǔ)元素的阻塞隊(duì)列。
LinkedTransferQueue：鏈表結(jié)構(gòu)組成的無(wú)界阻塞隊(duì)列。
LinkedBlockingDeque：鏈表結(jié)構(gòu)組成的雙向阻塞隊(duì)列。

阻塞隊(duì)列提供了下列四種處理方法:

這4類方法中，在隊(duì)列已滿（或?yàn)榭眨┑那闆r下，有些會(huì)拋出異常，有些則返回true/false，有些則一直阻塞，還有些則可以設(shè)置超時(shí)時(shí)間，時(shí)間到了后，自動(dòng)退出阻塞狀態(tài)，實(shí)際項(xiàng)目中可根據(jù)需要選取適合的方法。

2.隊(duì)列

隊(duì)列(Queue)與棧(Stack)是數(shù)據(jù)結(jié)構(gòu)中的二種常用結(jié)構(gòu)，隊(duì)列的特點(diǎn)是先進(jìn)先出(FirstInFirstOut)，而Stack是先進(jìn)后出(FirstInLastOut)，說(shuō)得通俗點(diǎn)：Queue就是電影院入場(chǎng)時(shí)人們排起來(lái)的進(jìn)場(chǎng)隊(duì)伍，先來(lái)的人（即：前排在前面的人）先入場(chǎng)，而Statck則是一隊(duì)人依次進(jìn)入了一個(gè)死胡同想出來(lái)，先進(jìn)去(最里面)的人，必須等后面的人(后進(jìn)入的人)出來(lái)了，自己才能出來(lái)。

阻塞隊(duì)列與普通隊(duì)列的區(qū)別在于，當(dāng)隊(duì)列是空的時(shí)，從隊(duì)列中獲取元素的操作將會(huì)被阻塞，或者當(dāng)隊(duì)列是滿時(shí)，往隊(duì)列里添加元素的操作會(huì)被阻塞。試圖從空的阻塞隊(duì)列中獲取元素的線程將會(huì)被阻塞，直到其他的線程往空的隊(duì)列插入新的元素。同樣，試圖往已滿的阻塞隊(duì)列中添加新元素的線程同樣也會(huì)被阻塞，直到其他的線程使隊(duì)列重新變得空閑起來(lái)，如從隊(duì)列中移除一個(gè)或者多個(gè)元素，或者完全清空隊(duì)列.

隊(duì)列在多線程應(yīng)用中，常用于生產(chǎn)-消費(fèi)場(chǎng)景，打個(gè)通俗的比方：很多人早上喜歡去買(mǎi)油條，買(mǎi)油條的人相當(dāng)于消費(fèi)者，做油條的師傅則是生產(chǎn)者。而油鍋邊上用于放油條的鐵架子，可以看成一個(gè)共享的隊(duì)列，師傅做好油條后，一根一根的撈出來(lái)放在架子上，而顧客則按排隊(duì)的順序一根根的付好錢(qián)從架子上拿。即：隊(duì)列的一頭，不斷有人在放入東西（生產(chǎn)元素），另一頭不斷有人的消費(fèi)(拿走元素)。這里就有一個(gè)很有趣的現(xiàn)象，如果買(mǎi)的人多，師傅來(lái)不及做，那么第一個(gè)顧客就會(huì)一直等著（后面的所有人也得等著，或稱為阻塞了后面的人），直到師傅炸好一根，然后第一個(gè)顧客買(mǎi)完走了，后面的人才能頂上來(lái)，類似的道理，如果架子放滿了，沒(méi)有人來(lái)買(mǎi)，師傅就會(huì)停下來(lái)，等有人來(lái)買(mǎi)了，才會(huì)繼續(xù)做，這就是所謂的隊(duì)列阻塞，而能產(chǎn)生阻塞行為的隊(duì)列稱為阻塞隊(duì)列。

從剛才的描述可以看出，發(fā)生阻塞起碼得滿足下面至少一個(gè)條件：(前提：隊(duì)列是有界的)

1.從隊(duì)列里取元素時(shí)，如果隊(duì)列為空，則代碼一直等在這里（即阻塞），直到隊(duì)列里有東西了，拿到元素了，后面的代碼才能繼續(xù)；

2.向隊(duì)列里放元素時(shí)，如果隊(duì)列滿了（即放不下更多元素），則代碼也會(huì)卡住，直到隊(duì)列里的東西被取走了（即：有空位可以放新元素了），后面的代碼才能繼續(xù)；

3.模擬生產(chǎn)者消費(fèi)者

模擬了買(mǎi)油條的場(chǎng)景，1個(gè)老板在做油條，3個(gè)顧客在排隊(duì)買(mǎi)：

package com.tl.skyLine.thread;
import java.util.concurrent.ArrayBlockingQueue;
/**
 * Created by tl on 17/3/3.
 */
public class BlockingQueueTest {
  //final成員變量表示常量，只能被賦值一次，賦值后值不再改變。
  private static final int queueSize = 5;
  private static final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(queueSize);
  private static final int produceSpeed = 2000;//生產(chǎn)速度(越小越快)
  private static final int consumeSpeed = 10;//消費(fèi)速度(越小越快)
  //生產(chǎn)者
  public static class Producer implements Runnable {
    @Override
    public void run() {
      while (true) {
        try {
          System.out.println("老板準(zhǔn)備炸油條了，架子上還能放：" + (queueSize - queue.size()) + "根油條");
          queue.put("1根油條");
          System.out.println("老板炸好了1根油條，架子上還能放：" + (queueSize - queue.size()) + "根油條");
          Thread.sleep(produceSpeed);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
  //消費(fèi)者
  public static class Consumer implements Runnable {
    @Override
    public void run() {
      while (true) {
        try {
          System.out.println("A 準(zhǔn)備買(mǎi)油條了，架子上還剩" + queue.size() + "根油條");
          queue.take();
          System.out.println("A 買(mǎi)到1根油條，架子上還剩" + queue.size() + "根油條");
          Thread.sleep(consumeSpeed);
          System.out.println("B 準(zhǔn)備買(mǎi)油條了，架子上還剩" + queue.size() + "根油條");
          queue.take();
          System.out.println("B 買(mǎi)到1根油條，架子上還剩" + queue.size() + "根油條");
          Thread.sleep(consumeSpeed);
          System.out.println("C 準(zhǔn)備買(mǎi)油條了，架子上還剩" + queue.size() + "根油條");
          queue.take();
          System.out.println("C 買(mǎi)到1根油條，架子上還剩" + queue.size() + "根油條");
          Thread.sleep(consumeSpeed);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
  public static void main(String[] args) {
    Thread producer = new Thread(new Producer());
    Thread consumer = new Thread(new Consumer());
    producer.start();
    consumer.start();
  }
}

結(jié)果：

老板準(zhǔn)備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
A 準(zhǔn)備買(mǎi)油條了，架子上還剩1根油條
A 買(mǎi)到1根油條，架子上還剩0根油條
B 準(zhǔn)備買(mǎi)油條了，架子上還剩0根油條
老板準(zhǔn)備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
B 買(mǎi)到1根油條，架子上還剩0根油條
C 準(zhǔn)備買(mǎi)油條了，架子上還剩0根油條
老板準(zhǔn)備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
C 買(mǎi)到1根油條，架子上還剩0根油條
A 準(zhǔn)備買(mǎi)油條了，架子上還剩0根油條
老板準(zhǔn)備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
A 買(mǎi)到1根油條，架子上還剩0根油條
B 準(zhǔn)備買(mǎi)油條了，架子上還剩0根油條
老板準(zhǔn)備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條

上面設(shè)置消費(fèi)速度快于生產(chǎn)速度，所有會(huì)發(fā)現(xiàn)控制臺(tái)中，老板炸油條時(shí)，架子上油條賣(mài)得快，所以能放4根或者5根油條，而顧客由于購(gòu)買(mǎi)速度快，所以會(huì)經(jīng)常出現(xiàn)架子上還剩0根油條，當(dāng)我們把生產(chǎn)速度produceSpeed跟消費(fèi)速度consumeSpeed對(duì)換一下，又會(huì)發(fā)現(xiàn)相反的結(jié)果。

4.原理分析

我們簡(jiǎn)單模擬一下BlockingQueue，熟悉一下背后的原理：

public class BlockingQueue {
	private List queue = new LinkedList();
	private int limit = 10;
	public BlockingQueue(int limit){
		this.limit = limit;
	}
	public synchronized void enqueue(Object item)
	throws InterruptedException {
		while(this.queue.size() == this.limit) {
			wait();
		}
		if(this.queue.size() == 0) {
			notifyAll();
		}
		this.queue.add(item);
	}
	public synchronized Object dequeue()
	throws InterruptedException{
		while(this.queue.size() == 0){
			wait();
		}
		if(this.queue.size() == this.limit){
			notifyAll();
		}
		return this.queue.remove(0);
	}
}

我們看一下jdk的實(shí)現(xiàn)過(guò)程：

ArrayBlockingQueue的部分源碼：

/** Main lock guarding all access */ 
final ReentrantLock lock; 
/** Condition for waiting takes */ 
private final Condition notEmpty; 
/** Condition for waiting puts */ 
private final Condition notFull;

這3個(gè)變量很重要，ReentrantLock重入鎖，notEmpty檢查不為空的Condition以及notFull用來(lái)檢查隊(duì)列未滿的Condition

Condition是一個(gè)接口，里面有二個(gè)重要的方法：

await()：Causesthecurrentthreadtowaituntilitissignalledorinterrupted.即阻塞當(dāng)前線程，直到被通知（喚醒）或中斷

singal():Wakesuponewaitingthread.喚醒阻塞的線程

再來(lái)看put方法：(jdk1.8)

public void put(E e) throws InterruptedException {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
    while (count == items.length)
      notFull.await();
    enqueue(e);
  } finally {
    lock.unlock();
  }
}

1.先獲取鎖
2.然后用while循環(huán)檢測(cè)元素個(gè)數(shù)是否等于items長(zhǎng)度，如果相等，表示隊(duì)列滿了,調(diào)用notFull的await()方法阻塞線程
3.否則調(diào)用enqueue()方法添加元素
4.最后解鎖

這是添加元素的代碼（jdk 1.8）,注意最后一行notEmpty.signal()方法，表示添加完元素后，調(diào)用singal()通知等待（從隊(duì)列中取元素）的線程，隊(duì)列不空(有值)啦，可以來(lái)取東西了。

類似的take()與dequeue()方法則相當(dāng)于逆過(guò)程（注：同樣都是jdk 1.8）

public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
    while (count == 0)
      notEmpty.await();
    return dequeue();
  } finally {
    lock.unlock();
  }
}

類似的：

1. 先加鎖
2. 如果元素個(gè)數(shù)為空，表示隊(duì)列已空，調(diào)用notEmpty的await()阻塞線程，直接隊(duì)列里又有新元素加入為止
3. 然后調(diào)用dequeue 從隊(duì)列里刪除元素
4. 解鎖

dequeue方法：

private E dequeue() {
  // assert lock.getHoldCount() == 1;
  // assert items[takeIndex] != null;
  final Object[] items = this.items;
  @SuppressWarnings("unchecked")
  E x = (E) items[takeIndex];
  items[takeIndex] = null;
  if (++takeIndex == items.length)
    takeIndex = 0;
  count--;
  if (itrs != null)
    itrs.elementDequeued();
  notFull.signal();
  return x;
}

倒數(shù)第2行，元素移除后，調(diào)用notFull.singnal喚醒等待（向隊(duì)列添加元素的）線程，隊(duì)列有空位了，可以向里面添加元素了。

總結(jié)

以上就是本文關(guān)于java并發(fā)學(xué)習(xí)之BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者詳解的全部?jī)?nèi)容，希望對(duì)大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站：Java并發(fā)之嵌套管程鎖死詳解、java并發(fā)編程之cas詳解、Java并發(fā)編程Semaphore計(jì)數(shù)信號(hào)量詳解等，有什么問(wèn)題可以隨時(shí)留言，小編會(huì)及時(shí)回復(fù)大家的。感謝朋友們對(duì)本站的支持！

您可能感興趣的文章: