怎麼利用Java手寫阻塞佇列

需求分析

阻塞佇列的主要的需求如下：

• ##佇列基礎的功能需要有，往隊列當中放數據，從佇列當中取數據。

• 所有的佇列操作都要是

  並發安全性的。

• 當佇列滿了之後再往隊列當中放資料的時候，執行緒需要被掛起，當佇列當中的資料被取出，讓佇列當中有空間的時候執行緒需要被喚醒。

• 當佇列空了之後再往佇列當中取資料的時候，執行緒需要被掛起，當有執行緒往佇列當中加入資料的時候被掛起的執行緒需要被喚醒。

• 在我們實現的隊列當中我們使用數組去存儲數據，因此在構造函數當中需要提供數組的初始大小，設置用多大的數組。

阻塞佇列實作原理

執行緒阻塞和喚醒

#在上面我們已經談到了阻塞佇列是

並發安全性的，而且我們也有將執行緒喚醒和阻塞的需求，因此我們可以選擇可重入鎖定ReentrantLock保證並發安全，但是我們還需要將執行緒喚醒和阻塞，因此我們可以選擇條件變數Condition進行執行緒的喚醒和阻塞操作，在Condition當中我們將會使用到的，主要有以下兩個函數：

• signal用來喚醒線程，當一個執行緒呼叫Condition的signal函數的時候就可以喚醒一個被await函數阻塞的執行緒。

• await用於阻塞線程，當一個線程呼叫Condition的await函數的時候這個線程就會阻塞。

陣列循環使用

因為佇列是一端進一端出，因此佇列肯定有頭有尾。

當我們在佇列當中加入一些資料之後，佇列的情況可能如下：

##在上圖中的基礎之上我們在進行四次出隊操作，結果如下：

在上面的狀態下，我們繼續加入8個數據，那麼佈局情況如下：

我們知道上圖在加入資料的時候不僅將陣列後半部的空間使用完了，而且可以繼續使用前半部沒有使用過的空間，也就是說在佇列內部實作了一個循環使用的過程。

為了確保數組的循環使用，我們需要用一個變數記錄隊列頭在數組當中的位置，用一個變量記錄隊列尾部在數組當中的位置，還需要有一個變量記錄隊列當中有多少個數據。

程式碼實作

成員變數定義

根據上面的分析我們可以知道，在我們自己實作的類別當中我們需要有如下的類別成員變數：

// 用于保护临界区的锁
private final ReentrantLock lock;
// 用于唤醒取数据的时候被阻塞的线程
private final Condition notEmpty;
// 用于唤醒放数据的时候被阻塞的线程
private final Condition notFull;
// 用于记录从数组当中取数据的位置 也就是队列头部的位置
private int takeIndex;
// 用于记录从数组当中放数据的位置 也就是队列尾部的位置
private int putIndex;
// 记录队列当中有多少个数据
private int count;
// 用于存放具体数据的数组
private Object[] items;

建構子

我們的建構子也很簡單，最核心的就是傳入一個陣列大小的參數，並且給上面的變數初始化賦值。

@SuppressWarnings("unchecked")
public MyArrayBlockingQueue(int size) {
  this.lock = new ReentrantLock();
  this.notEmpty = lock.newCondition();
  this.notFull = lock.newCondition();
  // 其实可以不用初始化 类会有默认初始化 默认初始化为0
  takeIndex = 0;
  putIndex = 0;
  count = 0;
  // 数组的长度肯定不能够小于0
  if (size <= 0)
    throw new RuntimeException("size can not be less than 1");
  items = (E[])new Object[size];
}

put函數

這是一個比較重要的函數了，在這個函數當中如果隊列沒有滿，則直接將資料放入到數組當中即可，如果數組滿了，則需要將線程掛起。

public void put(E x){
  // put 函数可能多个线程调用 但是我们需要保证在给变量赋值的时候只能够有一个线程
  // 因为如果多个线程同时进行赋值的话 那么可能后一个线程的赋值操作覆盖了前一个线程的赋值操作
  // 因此这里需要上锁
  lock.lock();
 
  try {
    // 如果队列当中的数据个数等于数组的长度的话 说明数组已经满了
    // 这个时候需要将线程挂起
    while (count == items.length)
      notFull.await(); // 将调用 await的线程挂起
    // 当数组没有满 或者在挂起之后再次唤醒的话说明数组当中有空间了
    // 这个时候需要将数组入队 
    // 调用入队函数将数据入队
    enqueue(x);
  } catch (InterruptedException e) {
    e.printStackTrace();
  } finally {
    // 解锁
    lock.unlock();
  }
}
 
// 将数据入队
private void enqueue(E x) {
  this.items[putIndex] = x;
  if (++putIndex == items.length)
    putIndex = 0;
  count++;
  notEmpty.signal(); // 唤醒一个被 take 函数阻塞的线程唤醒
}

offer函數

offer函數和put函數一樣，但是與put函數不同的是，當陣列當中資料填滿之後offer函數傳回

false

，而不是被阻塞。 <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:java;">public boolean offer(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { // 如果数组满了 则直接返回false 而不是被阻塞 if (count == items.length) return false; else { // 如果数组没有满则直接入队 并且返回 true enqueue(e); return true; } } finally { lock.unlock(); } }</pre><div class="contentsignin">登入後複製</div></div>add函數

這個函數和上面兩個函數作用一樣，也是往佇列當中加入數據，但當單一佇列滿了之後這個函數會拋出例外。

public boolean add(E e) {
  if (offer(e))
    return true;
  else
    throw new RuntimeException("Queue full");
}

take函數

這個函數主要是從佇列當中取出一個數據，但是當佇列為空的時候，這個函數會阻塞呼叫該函數的執行緒：

public E take() throws InterruptedException {
  // 这个函数也是不能够并发的 否则可能不同的线程取出的是同一个位置的数据
  // 进行加锁操作
  lock.lock();
  try {
    // 当 count 等于0 说明队列为空
    // 需要将线程挂起等待
    while (count == 0)
      notEmpty.await();
    // 当被唤醒之后进行出队操作
    return dequeue();
  }finally {
    lock.unlock();
  }
}
 
private E  dequeue() {
  final Object[] items = this.items;
  @SuppressWarnings("unchecked")
  E x = (E) items[takeIndex];
  items[takeIndex] = null; // 将对应的位置设置为 null GC就可以回收了
  if (++takeIndex == items.length)
    takeIndex = 0;
  count--; // 队列当中数据少一个了
  // 因为出队了一个数据 可以唤醒一个被 put 函数阻塞的线程 如果这个时候没有被阻塞的线程
  // 这个函数就不会起作用 也就说在这个函数调用之后被 put 函数挂起的线程也不会被唤醒
  notFull.signal(); // 唤醒一个被 put 函数阻塞的线程
  return x;
}

重寫toString函數

因為我們在後面的測試函數當中會列印我們這個類，而列印這個類別的時候會呼叫物件的

toString

方法得到一個字串，最後列印這個字串。 <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:java;">@Override public String toString() { StringBuilder stringBuilder = new StringBuilder(); stringBuilder.append(&quot;[&quot;); // 这里需要上锁 因为我们在打印的时候需要打印所有的数据 // 打印所有的数据就需要对数组进行遍历操作 而在进行遍历 // 操作的时候是不能进行插入和删除操作的 因为打印的是某 // 个时刻的数据 lock.lock(); try { if (count == 0) stringBuilder.append(&quot;]&quot;); else { int cur = 0; // 对数据进行遍历 一共遍历 count 次 因为数组当中一共有 count // 个数据 while (cur != count) { // 从 takeIndex 位置开始进行遍历 因为数据是从这个位置开始的 stringBuilder.append(items[(cur + takeIndex) % items.length].toString() + &quot;, &quot;); cur += 1; } // 删除掉最后一次没用的 &quot;, &quot; stringBuilder.delete(stringBuilder.length() - 2, stringBuilder.length()); stringBuilder.append(&amp;#39;]&amp;#39;); } }finally { lock.unlock(); } return stringBuilder.toString(); }</pre><div class="contentsignin">登入後複製</div></div>完整程式碼

整個我們自己完成的阻塞佇列的程式碼如下：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
 
public class MyArrayBlockingQueue<E> {
 
  // 用于保护临界区的锁
  private final ReentrantLock lock;
  // 用于唤醒取数据的时候被阻塞的线程
  private final Condition notEmpty;
  // 用于唤醒放数据的时候被阻塞的线程
  private final Condition notFull;
  // 用于记录从数组当中取数据的位置 也就是队列头部的位置
  private int takeIndex;
  // 用于记录从数组当中放数据的位置 也就是队列尾部的位置
  private int putIndex;
  // 记录队列当中有多少个数据
  private int count;
  // 用于存放具体数据的数组
  private Object[] items;
 
 
  @SuppressWarnings("unchecked")
  public MyArrayBlockingQueue(int size) {
    this.lock = new ReentrantLock();
    this.notEmpty = lock.newCondition();
    this.notFull = lock.newCondition();
    // 其实可以不用初始化 类会有默认初始化 默认初始化为0
    takeIndex = 0;
    putIndex = 0;
    count = 0;
    if (size <= 0)
      throw new RuntimeException("size can not be less than 1");
    items = (E[])new Object[size];
  }
 
  public void put(E x){
    lock.lock();
 
    try {
      while (count == items.length)
        notFull.await();
      enqueue(x);
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
  }
 
  private void enqueue(E x) {
    this.items[putIndex] = x;
    if (++putIndex == items.length)
      putIndex = 0;
    count++;
    notEmpty.signal();
  }
 
  private E  dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
      takeIndex = 0;
    count--;
    notFull.signal();
    return x;
  }
 
  public boolean add(E e) {
    if (offer(e))
      return true;
    else
      throw new RuntimeException("Queue full");
  }
 
  public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      if (count == items.length)
        return false;
      else {
        enqueue(e);
        return true;
      }
    } finally {
      lock.unlock();
    }
  }
 
  public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      return (count == 0) ? null : dequeue();
    } finally {
      lock.unlock();
    }
  }
 
  public E take() throws InterruptedException {
    lock.lock();
    try {
      while (count == 0)
        notEmpty.await();
      return dequeue();
    }finally {
      lock.unlock();
    }
  }
 
  @Override
  public String toString() {
    StringBuilder stringBuilder = new StringBuilder();
    stringBuilder.append("[");
    lock.lock();
    try {
      if (count == 0)
        stringBuilder.append("]");
      else {
        int cur = 0;
        while (cur != count) {
          stringBuilder.append(items[(cur + takeIndex) % items.length].toString()).append(", ");
          cur += 1;
        }
        stringBuilder.delete(stringBuilder.length() - 2, stringBuilder.length());
        stringBuilder.append(&#39;]&#39;);
      }
    }finally {
      lock.unlock();
    }
    return stringBuilder.toString();
  }
 
}

現在對上面的程式碼進行測試：

我們現在使用阻塞隊列模擬一個生產者消費者模型，設定阻塞隊列的大小為5，生產者線程會往隊列當中加入數據，數據為0-9的10個數字，消費者線程一共會消費10次。

import java.util.concurrent.TimeUnit;
 
public class Test {
 
  public static void main(String[] args) throws InterruptedException {
    MyArrayBlockingQueue<Integer> queue = new MyArrayBlockingQueue<>(5);
    Thread thread = new Thread(() -> {
      for (int i = 0; i < 10; i++) {
        System.out.println(Thread.currentThread().getName() + " 往队列当中加入数据：" + i);
        queue.put(i);
      }
    }, "生产者");
 
 
    Thread thread1 = new Thread(() -> {
      for (int i = 0; i < 10; i++) {
        try {
          System.out.println(Thread.currentThread().getName() + " 从队列当中取出数据：" + queue.take());
          System.out.println(Thread.currentThread().getName() + " 当前队列当中的数据：" + queue);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }, "消费者");
    thread.start();
    TimeUnit.SECONDS.sleep(3);
    thread1.start();
 
  }
}

上面程式碼的輸出如下所示：

生產者將資料加入佇列3
生產者將資料加入佇列當中：4
生產者往佇列當中加入資料：5
消費者從佇列當中取出資料：0
生產者往佇列當中加入資料：6
消費者目前隊列當中的資料：[1, 2, 3, 4, 5]
消費者從隊列當中取出資料：1
消費者目前隊列當中的資料：[2, 3, 4 , 5]
消費者從隊列當中取出資料：2
消費者目前隊列當中的資料：[3, 4, 5, 6]
生產者將資料加入資料在隊列中：7
消費者從隊列當中取出資料：3
消費者目前隊列當中的資料：[4, 5, 6, 7]
消費者從隊列當中取出資料：4
消費者目前隊列當中的資料：[5, 6, 7]
消費者從佇列當中取出資料：5
消費者目前佇列當中的資料：[6, 7]
生產者往佇列當中加入資料：8
消費者從佇列當中取出資料：6
消費者目前佇列當中的資料：[7, 8]
消費者從佇列當中取出資料：7
消費者目前佇列當中的資料： [8]
消費者從佇列當中取出資料：8
消費者目前佇列當中的資料：[]
生產者往佇列當中加入資料：9
消費者從佇列當中取出資料：9
消費者目前佇列當中的資料：[]


從上面的輸出結果我們知道，生產者執行緒列印5之後被掛起了，因為如果沒有被掛起，生產者線程肯定可以一次性輸出完成，因為消費者線程阻塞了3秒。由於阻塞佇列已滿，他在列印數字5後就未完成輸出，導致生產者線程被掛起。一旦消費者開始消費，阻塞隊列中就會騰出空間，生產者執行緒就可以繼續生產。

以上是怎麼利用Java手寫阻塞佇列的詳細內容。更多資訊請關注PHP中文網其他相關文章！