摘要：引言在包中，很好的解決了在多線程中，如何高效安全傳輸數(shù)據(jù)的問(wèn)題。同時(shí)，也用于自帶線程池的緩沖隊(duì)列中，了解也有助于理解線程池的工作模型。

在java.util.Concurrent包中，BlockingQueue很好的解決了在多線程中，如何高效安全“傳輸”數(shù)據(jù)的問(wèn)題。通過(guò)這些高效并且線程安全的隊(duì)列類，為我們快速搭建高質(zhì)量的多線程程序帶來(lái)極大的便利。同時(shí)，BlockingQueue也用于java自帶線程池的緩沖隊(duì)列中，了解BlockingQueue也有助于理解線程池的工作模型。

該接口屬于隊(duì)列，所以繼承了Queue接口，該接口最重要的五個(gè)方法分別是offer方法，poll方法，put方法，take方法和drainTo方法。

offer方法和poll方法分別有一個(gè)靜態(tài)重載方法，分別是offer(E e, long timeout, TimeUnit unit)和poll(long timeout, TimeUnit unit)方法。其意義是在限定時(shí)間內(nèi)存入或取出對(duì)象，如果不能存入取出則返回false。

put方法會(huì)在當(dāng)隊(duì)列存儲(chǔ)對(duì)象達(dá)到限定值時(shí)阻塞線程，而在隊(duì)列不為空時(shí)喚醒被take方法所阻塞的線程。take方法是相反的。

drainTo方法可批量獲取隊(duì)列中的元素。

LinkedBlockingQueue是比較常見(jiàn)的BlockingQueue的實(shí)現(xiàn)，他是基于鏈表的阻塞隊(duì)列。在創(chuàng)建該對(duì)象時(shí)如果不指定可存儲(chǔ)對(duì)象個(gè)數(shù)大小時(shí)，默認(rèn)為Integer.MAX_VALUE。當(dāng)生產(chǎn)者往隊(duì)列中放入一個(gè)數(shù)據(jù)時(shí)，隊(duì)列會(huì)從生產(chǎn)者手中獲取數(shù)據(jù)，并緩存在隊(duì)列內(nèi)部，而生產(chǎn)者立即返回；只有當(dāng)隊(duì)列緩沖區(qū)達(dá)到最大值緩存容量時(shí)，才會(huì)阻塞生產(chǎn)者隊(duì)列，直到消費(fèi)者從隊(duì)列中消費(fèi)掉一份數(shù)據(jù)，生產(chǎn)者線程會(huì)被喚醒，反之對(duì)于消費(fèi)者這端的處理也基于同樣的原理。

LinkedBlockingQueue內(nèi)部使用了獨(dú)立的兩把鎖來(lái)控制數(shù)據(jù)同步，這也意味著在高并發(fā)的情況下生產(chǎn)者和消費(fèi)者可以并行地操作隊(duì)列中的數(shù)據(jù)，以此來(lái)提高整個(gè)隊(duì)列的并發(fā)性能。

put方法和offer方法：

public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        int c = -1;
        Node node = new Node(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }
 
    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        int c = -1;
        Node node = new Node(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }
}

這兩個(gè)方法的區(qū)別是put方法在容量達(dá)到上限時(shí)會(huì)阻塞，而offer方法則會(huì)直接返回false。

ArrayBlockingQueue是基于數(shù)組的阻塞隊(duì)列，除了有一個(gè)定長(zhǎng)數(shù)組外，ArrayBlockingQueue內(nèi)部還保存著兩個(gè)整形變量，分別標(biāo)識(shí)著隊(duì)列的頭部和尾部在數(shù)組中的位置。ArrayBlockingQueue在生產(chǎn)者放入數(shù)據(jù)和消費(fèi)者獲取數(shù)據(jù)，都是共用同一個(gè)鎖對(duì)象，由此也意味著兩者無(wú)法真正并行運(yùn)行，這點(diǎn)尤其不同于LinkedBlockingQueue。
ArrayBlockingQueue和LinkedBlockingQueue間還有一個(gè)明顯的不同之處在于，前者在插入或刪除元素時(shí)不會(huì)產(chǎn)生或銷毀任何額外的對(duì)象實(shí)例，而后者則會(huì)生成一個(gè)額外的Node對(duì)象。

是一種沒(méi)有緩沖的阻塞隊(duì)列，在生產(chǎn)者put的同時(shí)必須要有一個(gè)消費(fèi)者進(jìn)行take，否則就會(huì)阻塞。聲明一個(gè)SynchronousQueue有兩種不同的方式。公平模式和非公平模式的區(qū)別:如果采用公平模式：SynchronousQueue會(huì)采用公平鎖，并配合一個(gè)FIFO隊(duì)列來(lái)阻塞多余的生產(chǎn)者和消費(fèi)者，從而體系整體的公平策略；但如果是非公平模式（SynchronousQueue默認(rèn)）：SynchronousQueue采用非公平鎖，同時(shí)配合一個(gè)LIFO隊(duì)列來(lái)管理多余的生產(chǎn)者和消費(fèi)者，而后一種模式，如果生產(chǎn)者和消費(fèi)者的處理速度有差距，則很容易出現(xiàn)饑渴的情況，即可能有某些生產(chǎn)者或者是消費(fèi)者的數(shù)據(jù)永遠(yuǎn)都得不到處理。

PriorityBlockingQueue是基于優(yōu)先級(jí)的阻塞隊(duì)列，該隊(duì)列不會(huì)阻塞生產(chǎn)者，只會(huì)阻塞消費(fèi)者。

DelayQueue隊(duì)列存儲(chǔ)的對(duì)象只有指定的延遲時(shí)間到了才能被取出，該隊(duì)列也不會(huì)阻塞生產(chǎn)者。

在處理多線程生產(chǎn)者消費(fèi)者問(wèn)題時(shí)的演示代碼：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
/**
 * Created by gavin on 15-8-30.
 */
public class BlockingQueueTest {
 
    public static void main(String[] args)
    {
        BlockingQueue queue = new ArrayBlockingQueue(1000);
        Thread p1 = new Thread(new Producer(queue),"producer1");
        Thread p2 = new Thread(new Producer(queue),"producer2");
        Thread c1 = new Thread(new Consumer(queue),"consumer1");
        Thread c2 = new Thread(new Consumer(queue),"consumer2");
 
        p1.start();
        p2.start();
        c1.start();
        c2.start();
    }
}
 
 
 
class Producer implements Runnable{
    private BlockingQueue queue;
 
    public Producer(BlockingQueue queue) {
        this.queue = queue;
    }
 
    public void run() {
        int i = 0;
        while (!Thread.currentThread().isInterrupted())
        {
            try {
                queue.put(Thread.currentThread().getName()+" product "+i);
            } catch (InterruptedException e) {
                System.err.println(Thread.currentThread().getName() + " error");
            }
            i++;
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
 
            }
        }
    }
}
class Consumer implements Runnable{
    private BlockingQueue queue;
 
    public Consumer(BlockingQueue queue) {
        this.queue = queue;
    }
 
    public void run() {
        int i = 0;
        while (!Thread.currentThread().isInterrupted())
        {
            try {
                String str = queue.take();
                System.out.println(str);
            } catch (InterruptedException e) {
 
            }
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
 
            }
        }
    }
}

BlockingQueue在并發(fā)編程中扮演著重要的角色，既可以自己用來(lái)解決生產(chǎn)者消費(fèi)者問(wèn)題，也用于java自帶線程池的緩沖隊(duì)列。

參考：
BlockingQueue

更多文章：http://blog.gavinzh.com