java并發(fā)之ArrayBlockingQueue詳細(xì)介紹
ArrayBlockingQueue是常用的線程集合，在線程池中也常常被當(dāng)做任務(wù)隊(duì)列來(lái)使用。使用頻率特別高。他是維護(hù)的是一個(gè)循環(huán)隊(duì)列（基于數(shù)組實(shí)現(xiàn)），循環(huán)結(jié)構(gòu)在數(shù)據(jù)結(jié)構(gòu)中比較常見，但是在源碼實(shí)現(xiàn)中還是比較少見的。
線程安全的實(shí)現(xiàn)
線程安全隊(duì)列，基本是離不開鎖的。ArrayBlockingQueue使用的是ReentrantLock，配合兩種Condition，實(shí)現(xiàn)了集合的線程安全操作。這里稍微說(shuō)一個(gè)好習(xí)慣，下面是成員變量的聲明。
|
1
2
3
4
5
6
7
8
9
|
private static final long serialVersionUID = -817911632652898426L; final Object[] items; int takeIndex; int putIndex; int count; final ReentrantLock lock; private final Condition notEmpty; private final Condition notFull; transient Itrs itrs = null; |
賦值的操作基本都是在構(gòu)造函數(shù)里做的。這樣有個(gè)好處，代碼執(zhí)行可控。成員變量的初始化也是會(huì)合并在構(gòu)造方法里執(zhí)行的，但是在執(zhí)行順序上需要好好斟酌，如果寫在構(gòu)造方法里初始化，則沒有相關(guān)問題。
阻塞隊(duì)列的常用場(chǎng)所就是生產(chǎn)者消費(fèi)者。一般都是生產(chǎn)者放入，消費(fèi)者從頭取數(shù)據(jù)。下面重點(diǎn)說(shuō)這兩個(gè)操作。
這兩個(gè)操作都是依靠鎖來(lái)保證線程安全的。
生產(chǎn)操作
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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(); } } |
put等放入操作，首先是獲取鎖，如果發(fā)現(xiàn)數(shù)據(jù)滿了，就通過notFull的condition，來(lái)阻塞線程。這里的條件判定一定是用while而不是if，多線程情況下，可以被喚醒后發(fā)現(xiàn)又滿了。
|
1
2
3
4
5
6
7
8
|
private void enqueue(E x) { final Object[] items = this.items; items[putIndex] = x; if (++putIndex == items.length) putIndex = 0; count++; notEmpty.signal(); } |
這個(gè)是入隊(duì)列的操作。首先獲取維護(hù)的數(shù)組。putindex就是放入操作的標(biāo)志。這個(gè)操作會(huì)一直加。達(dá)到預(yù)定的長(zhǎng)度后就變成0從頭開始計(jì)數(shù)。這樣插入的操作就是一個(gè)循環(huán)的操作了，count就是用來(lái)做計(jì)數(shù)的，作為能否插入數(shù)據(jù)的一個(gè)標(biāo)準(zhǔn)，插入數(shù)據(jù)后就通過notEmpty的condition發(fā)出一個(gè)信號(hào)喚醒消費(fèi)線程。
消費(fèi)操作
|
1
2
3
4
5
6
7
8
9
10
11
|
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } } |
消費(fèi)的方法也是這樣。先獲取鎖，然后進(jìn)行條件判斷，如果沒有數(shù)據(jù)，則阻塞線程。注意點(diǎn)和put一樣。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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--; if (itrs != null) itrs.elementDequeued(); notFull.signal(); return x;} |
取數(shù)據(jù)的時(shí)候，也依靠takeIndex，這是一個(gè)標(biāo)志，這個(gè)數(shù)值也會(huì)一直增加，表示取的第一個(gè)數(shù)據(jù)的位置。如果這個(gè)標(biāo)志走到最后，然后變成0，從頭再來(lái)。這樣保證取出的數(shù)據(jù)都是fifo的順序。刪除的時(shí)候如果發(fā)現(xiàn)迭代中，則會(huì)修改迭代器的遍歷。然后通過notFull的condition來(lái)喚醒生產(chǎn)線程。
移除操作
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
public boolean remove(Object o) { if (o == null) return false; final Object[] items = this.items; final ReentrantLock lock = this.lock; lock.lock(); try { if (count > 0) { final int putIndex = this.putIndex; int i = takeIndex; do { if (o.equals(items[i])) { removeAt(i); return true; } if (++i == items.length) i = 0; } while (i != putIndex); } return false; } finally { lock.unlock(); } } |
對(duì)于remove操作就比較麻煩了，首先獲取鎖之后，把兩個(gè)標(biāo)志位本地化，然后找到要?jiǎng)h除的元素的位置。調(diào)用removeAt，這里刪除需要對(duì)標(biāo)志位做改變。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
void removeAt(final int removeIndex) { final Object[] items = this.items; if (removeIndex == takeIndex) { items[takeIndex] = null; if (++takeIndex == items.length) takeIndex = 0; count--; if (itrs != null) itrs.elementDequeued(); } else { final int putIndex = this.putIndex; for (int i = removeIndex;;) { int next = i + 1; if (next == items.length) next = 0; if (next != putIndex) { items[i] = items[next]; i = next; } else { items[i] = null; this.putIndex = i; break; } } count--; if (itrs != null) itrs.removedAt(removeIndex); } notFull.signal(); } |
如果刪除的元素是位置和takeindex一樣。那就可以直接刪除，然后讓刪除標(biāo)志位向后移動(dòng)。如果不是，則從刪除的位置開始，進(jìn)行后面向前面的數(shù)據(jù)覆蓋的操作。直到遇到putindex的前一個(gè)位置。然后把那個(gè)位置的數(shù)據(jù)設(shè)置為null。并且把putindex的位置往前移動(dòng)一格，正在迭代的時(shí)候要?jiǎng)h除數(shù)據(jù)并且喚醒生產(chǎn)線程。
感謝閱讀，希望能幫助到大家，謝謝大家對(duì)本站的支持！
原文鏈接：https://my.oschina.net/xpbob/blog/830366
