java.util.concurrent包学习之阻塞队列

前言：

  阻塞队列是jdk5 java.util.concurrent包中五个模块之一，一般用于消费者、生产者的情景，与一般队列的不同就是阻塞队列内部加入了Lock和Condition可以在某个线程试图向队列添加元素而该队列已经满时，或者从队列移出元素而队列为空时，自动阻塞该线程。

阻塞队列的作用和使用场景

  所以在协调多个线程之间的合作时，阻塞队列是一个有用的工具。生产线程可以周期性地将中间结果存储在阻塞队列中，而消费线程移除中间结果并加以逻辑处理。阻塞队列最大的作用就是可以自动平衡负载，因为如果生产者线程集运行得比消费者线程集慢，消费者线程就会发生阻塞，反之也是一样。假设一个场景：我们写一个监控文件变化的程序，开一个线程获取变化了的文件名，拿到之后我们会做解析，查找数据库等等其他操作。如果我们对这两个操作分开统计它们的处理时间，会发现后面的解析需要的时间远远大于前面得到文件名的时间。如果这个线程每一秒轮询一次，而解析查询操作有可能会超出一秒，这样就会让大量的变化文件来不及处理。这个使用就需要用到阻塞线程了，可以使用生产线程来进行文件变化检测，而多个消费线程来进行文件处理操作。

JDK 7 提供了7个阻塞队列

**1、ArrayBlockingQueue **数组结构组成的有界阻塞队列。
此队列按照先进先出（FIFO）的原则对元素进行排序，但是默认情况下不保证线程公平的访问队列，即如果队列满了，那么被阻塞在外面的线程对队列访问的顺序是不能保证线程公平（即先阻塞，先插入）的。

2、LinkedBlockingQueue一个由链表结构组成的有界阻塞队列
此队列按照先出先进的原则对元素进行排序

3、PriorityBlockingQueue 支持优先级的无界阻塞队列

4、DelayQueue 支持延时获取元素的无界阻塞队列，即可以指定多久才能从队列中获取当前元素

5、SynchronousQueue不存储元素的阻塞队列，每一个put必须等待一个take操作，否则不能继续添加元素。并且他支持公平访问队列。

6、LinkedTransferQueue由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列，多了tryTransfer和transfer方法

transfer方法
  如果当前有消费者正在等待接收元素（take或者待时间限制的poll方法），transfer可以把生产者传入的元素立刻传给消费者。如果没有消费者等待接收元素，则将元素放在队列的tail节点，并等到该元素被消费者消费了才返回。

tryTransfer方法
  用来试探生产者传入的元素能否直接传给消费者。，如果没有消费者在等待，则返回false。和上述方法的区别是该方法无论消费者是否接收，方法立即返回。而transfer方法是必须等到消费者消费了才返回。

7、LinkedBlockingDeque链表结构的双向阻塞队列，优势在于多线程入队时，减少一半的竞争。

阻塞队列的方法

阻塞队列测试使用

public class BlockQueueTest {

    private static BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(10);

    /**
     * 消费者线程，每一秒消费一次
     */
  static   class Consumer implements  Runnable{

        BlockingQueue queue;
        public  Consumer(BlockingQueue queue){
            this.queue = queue;
        }
        public void run() {
            System.out.println("开始消费----");
            while (true){
                try {
                    Object poll = queue.take();
                    Thread.sleep(500);
                    System.out.println(Thread.currentThread().getName()+"消费"+(String)poll);
                } catch (InterruptedException e) {
                    //被中断，捕获后清除中断状态
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

   static class Producer implements Runnable{

        BlockingQueue queue;

        Producer(BlockingQueue queue){
            this.queue = queue;
        }

        public void run() {
            System.out.println("开始生产-------");
            while (true){
                try {
                    queue.put("test");
                    System.out.println("生产线程生产test");
                } catch (InterruptedException e) {
                    //捕获中断，清除中断状态
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
    public static void main(String[] args) {
        //开启生产线程
        new Thread(new Producer(blockingQueue)).start();
        //开启三个消费线程
        new Thread(new Consumer(blockingQueue)).start();
        new Thread(new Consumer(blockingQueue)).start();
        new Thread(new Consumer(blockingQueue)).start();
    }

参考

• 《Java核心技术》