java并發(fā)學習之BlockingQueue實現(xiàn)生產(chǎn)者消費者詳解

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

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

1.介紹

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

JDK7提供了以下7個阻塞隊列：

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

阻塞隊列提供了下列四種處理方法:

這4類方法中，在隊列已滿（或為空）的情況下，有些會拋出異常，有些則返回true/false，有些則一直阻塞，還有些則可以設置超時時間，時間到了后，自動退出阻塞狀態(tài)，實際項目中可根據(jù)需要選取適合的方法。

2.隊列

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

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

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

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

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

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

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

模擬了買油條的場景，1個老板在做油條，3個顧客在排隊買：

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;//消費速度(越小越快)
  //生產(chǎn)者
  public static class Producer implements Runnable {
    @Override
    public void run() {
      while (true) {
        try {
          System.out.println("老板準備炸油條了，架子上還能放：" + (queueSize - queue.size()) + "根油條");
          queue.put("1根油條");
          System.out.println("老板炸好了1根油條，架子上還能放：" + (queueSize - queue.size()) + "根油條");
          Thread.sleep(produceSpeed);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
  //消費者
  public static class Consumer implements Runnable {
    @Override
    public void run() {
      while (true) {
        try {
          System.out.println("A 準備買油條了，架子上還剩" + queue.size() + "根油條");
          queue.take();
          System.out.println("A 買到1根油條，架子上還剩" + queue.size() + "根油條");
          Thread.sleep(consumeSpeed);
          System.out.println("B 準備買油條了，架子上還剩" + queue.size() + "根油條");
          queue.take();
          System.out.println("B 買到1根油條，架子上還剩" + queue.size() + "根油條");
          Thread.sleep(consumeSpeed);
          System.out.println("C 準備買油條了，架子上還剩" + queue.size() + "根油條");
          queue.take();
          System.out.println("C 買到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();
  }
}

結果：

老板準備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
A 準備買油條了，架子上還剩1根油條
A 買到1根油條，架子上還剩0根油條
B 準備買油條了，架子上還剩0根油條
老板準備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
B 買到1根油條，架子上還剩0根油條
C 準備買油條了，架子上還剩0根油條
老板準備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
C 買到1根油條，架子上還剩0根油條
A 準備買油條了，架子上還剩0根油條
老板準備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條
A 買到1根油條，架子上還剩0根油條
B 準備買油條了，架子上還剩0根油條
老板準備炸油條了，架子上還能放：5根油條
老板炸好了1根油條，架子上還能放：4根油條

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

4.原理分析

我們簡單模擬一下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的實現(xiàn)過程：

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個變量很重要，ReentrantLock重入鎖，notEmpty檢查不為空的Condition以及notFull用來檢查隊列未滿的Condition

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

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

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

再來看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)檢測元素個數(shù)是否等于items長度，如果相等，表示隊列滿了,調(diào)用notFull的await()方法阻塞線程
3.否則調(diào)用enqueue()方法添加元素
4.最后解鎖

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

類似的take()與dequeue()方法則相當于逆過程（注：同樣都是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. 如果元素個數(shù)為空，表示隊列已空，調(diào)用notEmpty的await()阻塞線程，直接隊列里又有新元素加入為止
3. 然后調(diào)用dequeue 從隊列里刪除元素
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喚醒等待（向隊列添加元素的）線程，隊列有空位了，可以向里面添加元素了。

總結

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

您可能感興趣的文章: