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

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

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

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

put方法會在當隊列存儲對象達到限定值時阻塞線程，而在隊列不為空時喚醒被take方法所阻塞的線程。take方法是相反的。

drainTo方法可批量獲取隊列中的元素。

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

LinkedBlockingQueue內部使用了獨立的兩把鎖來控制數據同步，這也意味著在高并發(fā)的情況下生產者和消費者可以并行地操作隊列中的數據，以此來提高整個隊列的并發(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;
    }
}

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

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

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

PriorityBlockingQueue是基于優(yōu)先級的阻塞隊列，該隊列不會阻塞生產者，只會阻塞消費者。

DelayQueue隊列存儲的對象只有指定的延遲時間到了才能被取出，該隊列也不會阻塞生產者。

在處理多線程生產者消費者問題時的演示代碼：

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ā)編程中扮演著重要的角色，既可以自己用來解決生產者消費者問題，也用于java自帶線程池的緩沖隊列。

參考：
BlockingQueue

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