今天重溫了下 java 多線程中的 notify() 方法以及 wait() 方法，一時(shí)興起，決定通過(guò)這倆個(gè)方法，實(shí)現(xiàn)一個(gè)簡(jiǎn)易的自定義阻塞隊(duì)列。

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

1.新建一個(gè) MyQueue.java 類(lèi)

import java.util.LinkedList;import java.util.concurrent.atomic.AtomicInteger;import com.xiaoleilu.hutool.util.StrUtil;/** * 使用 notify() 和 wait() 實(shí)現(xiàn)自定義阻塞隊(duì)列 * * @author Yangkai.Shen * @version 1.0 * @date 2017.08.02 at 11:51:14 */public class MyQueue {// 1. 承載數(shù)據(jù)的容器private LinkedList<Object> queue = new LinkedList<Object>();// 2. 計(jì)數(shù)器，用于判定邊界private AtomicInteger count = new AtomicInteger(0);private final int minSize = 0;// 3. 初始化一個(gè)對(duì)象，用于加鎖private final Object lock = new Object();private final int maxSize;public MyQueue(int maxSize) {this.maxSize = maxSize;}/** * 添加一個(gè)元素到隊(duì)列中，如果隊(duì)列元素已滿(mǎn)，則調(diào)用此方法的線程被阻塞，直到存在多余空間了，再進(jìn)行添加 * * @param obj 添加 obj 到隊(duì)列尾部 */public void put(Object obj) {synchronized (lock) {// 1.沒(méi)有多余空間，就阻塞線程while (count.get() == this.maxSize) {try {lock.wait();} catch (InterruptedException e) {e.printStackTrace();}}// 2.添加元素queue.add(obj);// 3.計(jì)數(shù)器累加count.incrementAndGet();System.out.println(StrUtil.format('新加入的元素為：{}', obj));// 4.喚醒其他線程（若本來(lái)元素為空，有線程調(diào)用 get 方法，那么原本被阻塞的，需要在此時(shí)被喚醒）lock.notify();}}/** * 獲取一個(gè)元素，如果隊(duì)列元素為空，則調(diào)用此方法的線程被阻塞，直到添加新元素了，再進(jìn)行獲取 * * @return 返回隊(duì)列的第一個(gè)元素 */public Object get() {Object ret = null;synchronized (lock) {// 1.沒(méi)有元素，就阻塞線程while (count.get() == this.minSize) {try {lock.wait();} catch (InterruptedException e) {e.printStackTrace();}}// 2.取第一個(gè)元素ret = queue.removeFirst();// 3.計(jì)數(shù)器遞減count.decrementAndGet();System.out.println(StrUtil.format('移除的元素為：{}', ret));// 4.喚醒其他線程（若元素本來(lái)已滿(mǎn)，有線程調(diào)用 put 方法，那么原本被阻塞的，需要在此時(shí)被喚醒）lock.notify();}return ret;}public int getSize() {return this.count.get();}}

2.新建一個(gè)測(cè)試類(lèi) MyQueueTest.java，測(cè)試類(lèi)中，我們初始化一個(gè)隊(duì)列，并將元素填滿(mǎn)，然后啟動(dòng)一個(gè)線程 t1，去插入數(shù)據(jù)，中間休眠 2s，再去啟動(dòng)一個(gè)線程 t2 取數(shù)據(jù)。

import com.xiaoleilu.hutool.util.StrUtil;import java.util.concurrent.TimeUnit;public class MyQueueTest {public static void main(String[] args) {final MyQueue queue = new MyQueue(5);queue.put('a');queue.put('b');queue.put('c');queue.put('d');queue.put('e');System.out.println(StrUtil.format('當(dāng)前隊(duì)列的長(zhǎng)度: {}', queue.getSize()));Thread t1 = new Thread(() -> {queue.put('f');queue.put('g');queue.put('h');}, 't1'); Thread t2 = new Thread(() -> { queue.get(); queue.get(); }); t1.start(); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } t2.start();}

3.啟動(dòng)測(cè)試類(lèi)，查看運(yùn)行結(jié)果。控制臺(tái)如果應(yīng)該出現(xiàn)的效果是，隊(duì)列先初始化完成，然后休眠 2s，接下來(lái)先取數(shù)據(jù)，再插入數(shù)據(jù)，則證明阻塞隊(duì)列生效。下面是控制臺(tái)運(yùn)行的效果：

初始化隊(duì)列

休眠 2s 后取隊(duì)首元素，再插入隊(duì)尾元素

此時(shí)我們會(huì)發(fā)現(xiàn)，程序還未停止，因?yàn)榇藭r(shí)隊(duì)列已滿(mǎn)，但是線程 t1 還未插入 h 元素，因此線程被阻塞著，直至下次隊(duì)列有空余空間才會(huì)被喚醒。

4.至此，一個(gè)自定義阻塞隊(duì)列就已經(jīng)實(shí)現(xiàn)了。

5.細(xì)心的朋友會(huì)發(fā)現(xiàn)，我打印的 log 里用到了一個(gè) StrUtil.format() 方法，這個(gè)和 slf4j 的 log 用法一致，可以使用占位符。這個(gè)是用到了一個(gè)國(guó)產(chǎn)良心工具類(lèi)，hutool，國(guó)產(chǎn)開(kāi)源，需要大家的支持，覺(jué)得好用的話，期望可以去 碼云 或者 github 上給個(gè) Star 吧！

以上就是Java實(shí)現(xiàn)自定義阻塞隊(duì)列的詳細(xì)內(nèi)容，更多關(guān)于Java 自定義阻塞隊(duì)列的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！