java多线程（九）阻塞队列

转载请注明出处：/article/1364855.html

前边的博客中我们介绍了如果用对象锁和条件锁以及更加方便的synchronized关键字来实现多线程的同步和互斥，也许你会觉得使用synchronized关键字已经非常方便了，但是使用者必须真正的理解synchronized的用法，而且要有一定的多线程的编程的经验，否则很难做到全面的考虑问题而造成意想不到的问题。其实在java中还有比synchronized关键字更加方便的途径来实现同步和互斥，并且不需要考虑复杂的问题，那就是使用阻塞队列，这一次我们就来讨论一下阻塞队列的使用。

例如我们前边经常举的例子——银行转账。我们每次写这个程序，都要考虑一大堆问题，例如两个线程之间应该怎么互斥，金额不足了怎么办等等。其实我们可以通过一个队列优雅且安全的解决这个问题。转账线程将转账指令对象插入到一个队列中，而不是直接访问银行对象。另一个线程从队列中取出指令执行转账。因为只有该线程可以访问该银行对象的内部方法，因此不需要同步和互斥。

阻塞队列是指，当试图向队列中添加元素而队列已满时，或者想从队列中移出一个元素而队列为空时，执行该操作的线程会被阻塞。在协调多个线程工作时，阻塞队列是一个有用的工具，工作者线程可以周期的将中间结果存储在队列中，其他的线程移出中间结果并进行进一步的处理。队列会进行自动的负载平衡，当队列满时，向队列中添加元素的线程会被阻塞，当队列为空时，试图从队列中取出元素的线程会被阻塞。

java中主要有以下几种阻塞队列：

1、ArrayBlockingQueue

这种队列用循环数组实现的，有两种构造方法：

1）带有指定容量的阻塞队列

ArrayBlockingQueue(int capacity)

2）带有指定容量和公平性设置的阻塞队列

ArrayBlockingQueue(int capacity,boolean fair)

公平性的阻塞队列在唤醒时会优先考虑等待时间最长的线程，但是这并不保证绝对的公平，因为公平的阻塞队列会降低效率，所以通常不采用，默认情况下是非公平的阻塞队列。

2、LinkedBlockingQueue

这种队列使用链表实现，有以下两种构造方法：

1）构造一个无上限的阻塞队列

LinkedBlockingQueue();

2）构造一个带上限的阻塞队列

LinkedBlockingQueue(int capacity);

3、LinkedBlockingDeque

这种队列的构造方法和实现原理都可第二种一样，他们两个之间的区别就是LinkedBlockingDeque是一种双向的队列。

4、DelayQueue

构造一个包含Delayed元素的无界的阻塞时间有限的阻塞队列。只有那些延时超多时间的元素可以从队列中移走。

构造方法：

DelayQueue();

5、PriorityBlockingQueue

优先级阻塞队列，优先级高的先从队列中移出。这种队列使用堆实现，有以下三种构造方法：

PriorityBlockingQueue();
PriorityBlockingQueue(int initialCapacity);
PriorityBlockingQueue(int initialCapacity,Comparator<? super E> compaator);

initialCapacity:优先队列的初始容量。默认值是11.

comparator 用来对元素进行比较的比较器，如果没有指定，则元素必须实现Comparable接口。

阻塞队列的常用方法

方法	正常动作	特殊情况下的动作
add	添加一个元素	如果队列为满，则抛出IllegalStateException异常
remove	移出并返回头元素	如果队列为空，则抛出NoSuchElementException异常
put	添加一个元素	如果队列为满，则阻塞
take	移出并返回头元素	如果队列为空，则阻塞
peek	返回队列的头元素	如果队列为空，则返回null
poll	移出并返回队列的头元素	如果队列为空，则返回null
element	返回队列的头元素	如果队列为空，则抛出NoSuchElementException异常
offer	添加一个元素并返回true	如果队列满，则返回false

使用阻塞队列实现生产者消费者问题：

生产者消费者问题是经典的线程同步问题，生产者和消费者共用一个缓冲区，可以看做仓库，生产者生产产品放入仓库，消费者从仓库中取出产品进行消费，当仓库为空时，消费者阻塞，当仓库为满，生产者阻塞。

import java.util.concurrent.ArrayBlockingQueue;

public class ProductThread implements Runnable {

    private ArrayBlockingQueue<Integer> queue;

    public ProductThread(ArrayBlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    public void run(){
        while(true){
            try {
                int product = (int)(Math.random()*10);
                /*
                 * 队列满时会阻塞
                 */
                queue.put(product);
                System.out.println(Thread.currentThread()+"生产了产品——"+product+" 剩余元素："+queue.size());
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }
}

import java.util.concurrent.ArrayBlockingQueue;

public class ConsumeThread implements Runnable {

    private ArrayBlockingQueue<Integer> queue;

    public ConsumeThread(ArrayBlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    public void run(){
        while(true){
            try {
                /*
                 * 队列空时会阻塞
                 */
                int product = queue.take();
                System.out.println(Thread.currentThread()+"消费了元素——"+product+" 剩余空间："+queue.size());
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }
}

import java.util.concurrent.ArrayBlockingQueue;

public class Main {

    public static void main(String[] args) {
        //大小为10的循环数组阻塞队列
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

        /*
         * 用5个生产者线程来生产产品
         */
        for(int i=0;i<5;i++){
            new Thread(new ProductThread(queue)).start();
        }

        /*
         * 用2个消费者线程来消费产品
         */
        for(int i=0;i<2;i++){
            new Thread(new ConsumeThread(queue)).start();
        }
    }

}

运行结果：

Thread[Thread-2,5,main]生产了产品——4 剩余元素：4

Thread[Thread-5,5,main]消费了元素——1 剩余空间：3

Thread[Thread-1,5,main]生产了产品——1 剩余元素：3

Thread[Thread-6,5,main]消费了元素——1 剩余空间：3

Thread[Thread-0,5,main]生产了产品——0 剩余元素：3

Thread[Thread-4,5,main]生产了产品——1 剩余元素：3

Thread[Thread-3,5,main]生产了产品——9 剩余元素：4

Thread[Thread-0,5,main]生产了产品——2 剩余元素：5

Thread[Thread-1,5,main]生产了产品——9 剩余元素：5

Thread[Thread-4,5,main]生产了产品——6 剩余元素：7

Thread[Thread-2,5,main]生产了产品——2 剩余元素：7

Thread[Thread-3,5,main]生产了产品——1 剩余元素：7

Thread[Thread-5,5,main]消费了元素——4 剩余空间：6

Thread[Thread-6,5,main]消费了元素——0 剩余空间：7

Thread[Thread-0,5,main]生产了产品——0 剩余元素：7

Thread[Thread-4,5,main]生产了产品——7 剩余元素：8

Thread[Thread-3,5,main]生产了产品——5 剩余元素：9

Thread[Thread-1,5,main]生产了产品——4 剩余元素：10

Thread[Thread-5,5,main]消费了元素——9 剩余空间：9

Thread[Thread-2,5,main]生产了产品——9 剩余元素：10

Thread[Thread-6,5,main]消费了元素——2 剩余空间：9

Thread[Thread-1,5,main]生产了产品——4 剩余元素：10

Thread[Thread-0,5,main]生产了产品——3 剩余元素：10

Thread[Thread-6,5,main]消费了元素——2 剩余空间：9

Thread[Thread-4,5,main]生产了产品——5 剩余元素：10

Thread[Thread-5,5,main]消费了元素——9 剩余空间：10

Thread[Thread-3,5,main]生产了产品——9 剩余元素：10

Thread[Thread-5,5,main]消费了元素——6 剩余空间：10

Thread[Thread-6,5,main]消费了元素——1 剩余空间：10

Thread[Thread-2,5,main]生产了产品——2 剩余元素：10

Thread[Thread-0,5,main]生产了产品——1 剩余元素：10

Thread[Thread-5,5,main]消费了元素——0 剩余空间：10

Thread[Thread-4,5,main]生产了产品——4 剩余元素：10

Thread[Thread-6,5,main]消费了元素——7 剩余空间：10

Thread[Thread-5,5,main]消费了元素——5 剩余空间：10

从结果上我们可以看出，刚开始阻塞队列为空，多个生产线程同时生产，当有了产品之后，两个消费者开始消费，因为生产者的数量较多，所以队列逐渐填满当队列满时，生产者线程阻塞，只能等待消费者消费一个产品后再生产，所以最后出现消费一个产品，生产一个产品的结果。

源码下载：http://download.csdn.net/detail/xingjiarong/9052057