快捷導(dǎo)航

Java多線程 BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型詳解

更新時(shí)間：2019年09月06日 08:59:39 作者：Rest探路者

這篇文章主要介紹了Java多線程 BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型詳解,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

BlockingQueue

BlockingQueue、解決了多線程中，如何高效安全“傳輸”數(shù)據(jù)的問題。程序員無需關(guān)心什么時(shí)候阻塞線程，什么時(shí)候喚醒線程，該喚醒哪個(gè)線程。

方法介紹

BlockingQueue是Queue的子類

void put(E e)

插入指定元素，當(dāng)BlockingQueue為滿，則線程阻塞，進(jìn)入Waiting狀態(tài)，直到BlockingQueue有空閑空間再繼續(xù)。
這里以ArrayBlockingQueue為例進(jìn)行分析

void take()

隊(duì)首出隊(duì)，當(dāng)BlockingQueue為空，則線程阻塞，進(jìn)入Waiting狀態(tài)，直到BlockingQueue不為空再繼續(xù)。

int drainTo(Collection<? super E> c)

從隊(duì)列中批量取出數(shù)據(jù)，并放入到另一個(gè)集合中，返回轉(zhuǎn)移數(shù)據(jù)的數(shù)量，只需一次加鎖和解鎖。

BlockingQueue的實(shí)現(xiàn)類

ArrayBlockingQueue

  /*
   * Concurrency control uses the classic two-condition algorithm
   * found in any textbook.
   */

  /** Main lock guarding all access */
  final ReentrantLock lock;

  /** Condition for waiting takes */
  private final Condition notEmpty;

  /** Condition for waiting puts */
  private final Condition notFull;

基于數(shù)組實(shí)現(xiàn)的BlockingQueue，需要指定隊(duì)列容量，可以指定是否為公平鎖；只有一個(gè)ReentrantLock，生產(chǎn)者和消費(fèi)者不能異步執(zhí)行。

LinkedBlockingQueue

  /** Lock held by take, poll, etc */
  private final ReentrantLock takeLock = new ReentrantLock();

  /** Wait queue for waiting takes */
  private final Condition notEmpty = takeLock.newCondition();

  /** Lock held by put, offer, etc */
  private final ReentrantLock putLock = new ReentrantLock();

  /** Wait queue for waiting puts */
  private final Condition notFull = putLock.newCondition();

基于鏈表實(shí)現(xiàn)的BlockingQueue，可以指定隊(duì)列容量，不指定隊(duì)列容量默認(rèn)為Integer.MAX_VALUE；有兩個(gè)ReentrantLock，生產(chǎn)者和消費(fèi)者可以異步執(zhí)行。

BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型

緩沖區(qū)可以存放大量數(shù)據(jù)

生產(chǎn)者和消費(fèi)者速度各不相同

public class MyThread42 {
  public static void main(String[] args)
  {
    final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
    Runnable producerRunnable = new Runnable()
    {
      int i = 0;
      public void run()
      {
        while (true)
        {
          try
          {
            System.out.println("我生產(chǎn)了一個(gè)" + i++);
            bq.put(i + "");
            Thread.sleep(1000);
          }
          catch (InterruptedException e)
          {
            e.printStackTrace();
          }
        }
      }
    };
    Runnable customerRunnable = new Runnable()
    {
      public void run()
      {
        while (true)
        {
          try
          {
            System.out.println("我消費(fèi)了一個(gè)" + bq.take());
            Thread.sleep(3000);
          }
          catch (InterruptedException e)
          {
            e.printStackTrace();
          }
        }
      }
    };
    Thread producerThread = new Thread(producerRunnable);
    Thread customerThread = new Thread(customerRunnable);
    producerThread.start();
    customerThread.start();
  }
}

輸出結(jié)果如下

我生產(chǎn)了一個(gè)0
我消費(fèi)了一個(gè)1
我生產(chǎn)了一個(gè)1
我生產(chǎn)了一個(gè)2
我消費(fèi)了一個(gè)2
我生產(chǎn)了一個(gè)3
我生產(chǎn)了一個(gè)4
我生產(chǎn)了一個(gè)5
我消費(fèi)了一個(gè)3
我生產(chǎn)了一個(gè)6
我生產(chǎn)了一個(gè)7
我生產(chǎn)了一個(gè)8
我消費(fèi)了一個(gè)4
我生產(chǎn)了一個(gè)9
我生產(chǎn)了一個(gè)10
我生產(chǎn)了一個(gè)11
我消費(fèi)了一個(gè)5
我生產(chǎn)了一個(gè)12
我生產(chǎn)了一個(gè)13
我生產(chǎn)了一個(gè)14
我消費(fèi)了一個(gè)6
我生產(chǎn)了一個(gè)15
我生產(chǎn)了一個(gè)16
我消費(fèi)了一個(gè)7
我生產(chǎn)了一個(gè)17
我消費(fèi)了一個(gè)8
我生產(chǎn)了一個(gè)18
我消費(fèi)了一個(gè)9
我生產(chǎn)了一個(gè)19
我消費(fèi)了一個(gè)10
我生產(chǎn)了一個(gè)20
我消費(fèi)了一個(gè)11
我生產(chǎn)了一個(gè)21
我消費(fèi)了一個(gè)12
我生產(chǎn)了一個(gè)22
我消費(fèi)了一個(gè)13
我生產(chǎn)了一個(gè)23
我消費(fèi)了一個(gè)14
我生產(chǎn)了一個(gè)24

······

生產(chǎn)者沒有生產(chǎn)到BlockingQueue的容量（極限是10）之前，生產(chǎn)3個(gè)，消費(fèi)1個(gè)，再生產(chǎn)到BlockingQueue的容量之后，生產(chǎn)一個(gè)消費(fèi)一個(gè)，因?yàn)椴荒艹^BlockingQueue的容量。

以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: