1. ArrayBlockingQueue, LinkedBlockingQueue, ConcurrentLinkedQueue

ArrayBlockingQueue, LinkedBlockingQueue 繼承自 BlockingQueue, 他們的特點(diǎn)就是 Blocking, Blocking 特有的方法就是 take() 和 put(), 這兩個(gè)方法是阻塞方法, 每當(dāng)隊(duì)列容量滿的時(shí)候, put() 方法就會(huì)進(jìn)入wait, 直到隊(duì)列空出來(lái), 而每當(dāng)隊(duì)列為空時(shí), take() 就會(huì)進(jìn)入等待, 直到隊(duì)列有元素可以 take()

ArrayBlockingQueue, LinkedBlockingQueue 區(qū)別在于 ArrayBlockingQueue 必須指定容量, 且可以指定 fair 變量, 如果 fair 為 true, 則會(huì)保持 take() 或者 put() 操作時(shí)線程的 block 順序, 先 block 的線程先 take() 或 put(), fair 又內(nèi)部變量 ReentrantLock 保證

ConcurrentLinkedQueue 通過(guò) CAS 操作實(shí)現(xiàn)了無(wú)鎖的 poll() 和 offer(), 他的容量是動(dòng)態(tài)的, 由于無(wú)鎖, 所以在 poll() 或者 offer() 的時(shí)候 head 與 tail 可能會(huì)改變, 所以它會(huì)持續(xù)的判斷 head 與 tail 是否改變來(lái)保證操作正確性, 如果改變, 則會(huì)重新選擇 head 與 tail. 而由于無(wú)鎖的特性, 他的元素更新與 size 變量更新無(wú)法做到原子 (實(shí)際上它沒有 size 變量), 所以他的 size() 是通過(guò)遍歷 queue 來(lái)獲得的, 在效率上是 O(n), 而且無(wú)法保證準(zhǔn)確性, 因?yàn)楸闅v的時(shí)候有可能 queue size 發(fā)生了改變.

?

RingBuffer 是 Distruptor 中的一個(gè)用來(lái)替代 ArrayBlockingQueue 的隊(duì)列, 它的思想在于長(zhǎng)度可控, 且無(wú)鎖, 只有在 blocking 的時(shí)候(沒有數(shù)據(jù)的時(shí)候出隊(duì), 數(shù)據(jù)滿的時(shí)候入隊(duì))會(huì)自旋. 實(shí)現(xiàn)原理是使用一個(gè)環(huán)形array, 生產(chǎn)者作為 tail, 消費(fèi)者作為 head, 每生產(chǎn)一次 tail atomic++, 每消費(fèi)一次 head atomic++, tail 不能超過(guò) head 一圈(array size, 即隊(duì)列滿時(shí) blocking), tail 不能超過(guò)自己tail一圈(即不能覆蓋未被消費(fèi)的值), head 不能超過(guò) tail (即無(wú)可消費(fèi)任務(wù)時(shí) blocking), head 不能取到空值(取到空值時(shí) blocking). blocking 使用一個(gè) while 自旋來(lái)完成, 那么只要生產(chǎn)者消費(fèi)者的速度相當(dāng)時(shí), 即可通過(guò) atomicInteger(cas) 保證無(wú)鎖, 而如果你需要在 blocking 的時(shí)候立即返回, 則 while 自旋都可以不需要. 相比于 ArrayBlockingQueue, 它可以絕大部分時(shí)間無(wú)鎖, blocking 自旋, 相比于 concurrentLinkedQueue, 他又能做到長(zhǎng)度限制. 代碼如下:

public class RingBuffer<T> implements Serializable {/****/private static final long serialVersionUID = 6976960108708949038L;private volatile AtomicInteger head;private volatile AtomicInteger tail;private int length;final T EMPTY = null;private volatile T[] queue;public RingBuffer(Class<T> type, int length){this.head = new AtomicInteger(0);this.tail = new AtomicInteger(0);this.length = length == 0 ? 2 << 16 : length; // 默認(rèn)2^16 this.queue = (T[]) Array.newInstance(type, this.length);}public void enQueue(T t){if(t == null) t= (T) new Object();// 阻塞 -- 避免多生成者循環(huán)生產(chǎn)同一個(gè)節(jié)點(diǎn) while(this.getTail() - this.getHead() >= this.length);int ctail = this.tail.getAndIncrement();while(this.queue[this.getTail(ctail)] != EMPTY); // 自旋 this.queue[this.getTail(ctail)] = t;}public T deQueue(){T t = null;// 阻塞 -- 避免多消費(fèi)者循環(huán)消費(fèi)同一個(gè)節(jié)點(diǎn) while(this.head.get() >= this.tail.get());int chead = this.head.getAndIncrement();while(this.queue[this.getHead(chead)] == EMPTY); // 自旋 t = this.queue[this.getHead(chead)];this.queue[this.getHead(chead)] = EMPTY;return t;}public int getHead(int index){return index & (this.length - 1);}public int getTail(int index) {return index & (this.length - 1);}public int getHead() {return head.get() & (this.length - 1);}public int getTail() {return tail.get() & (this.length - 1);}public T[] getQueue() {return queue;}public int getLength() {return length;}public void setLength(int length) {this.length = length;}}

?

?

轉(zhuǎn)載于:https://www.cnblogs.com/zemliu/p/3823597.html

總結(jié)

以上是生活随笔為你收集整理的ArrayBlockingQueue, LinkedBlockingQueue, ConcurrentLinkedQueue, RingBuffer的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。