阻塞佇列的主要的需求如下：
並發安全性的。
並發安全性的，而且我們也有將執行緒喚醒和阻塞的需求，因此我們可以選擇可重入鎖定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函數
，而不是被阻塞。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(); } }登入後複製
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函數
方法得到一個字串，最後列印這個字串。@Override public String toString() { StringBuilder stringBuilder = new StringBuilder(); stringBuilder.append("["); // 这里需要上锁 因为我们在打印的时候需要打印所有的数据 // 打印所有的数据就需要对数组进行遍历操作 而在进行遍历 // 操作的时候是不能进行插入和删除操作的 因为打印的是某 // 个时刻的数据 lock.lock(); try { if (count == 0) stringBuilder.append("]"); else { int cur = 0; // 对数据进行遍历 一共遍历 count 次 因为数组当中一共有 count // 个数据 while (cur != count) { // 从 takeIndex 位置开始进行遍历 因为数据是从这个位置开始的 stringBuilder.append(items[(cur + takeIndex) % items.length].toString() + ", "); cur += 1; } // 删除掉最后一次没用的 ", " stringBuilder.delete(stringBuilder.length() - 2, stringBuilder.length()); stringBuilder.append(']'); } }finally { lock.unlock(); } return stringBuilder.toString(); }登入後複製
import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; public class MyArrayBlockingQueue{ // 用于保护临界区的锁 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(']'); } }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 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中文網其他相關文章！
