1.wait()和notify()方式(摘自:https://segmentfault.com/a/1190000000373535)
 
 
阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列。
 
线程1往阻塞队列中添加元素，而线程2从阻塞队列中移除元素。
 
具体实现:
 

public class BlockingQueue {

  private List queue = new LinkedList();
  private int  limit = 10;

  public BlockingQueue(int limit){
    this.limit = limit;
  }


  public synchronized void enqueue(Object item)
  throws InterruptedException  {
    while(this.queue.size() == this.limit) {
      wait();
    }
    if(this.queue.size() == 0) {
      notifyAll();
    }
    this.queue.add(item);
  }


  public synchronized Object dequeue()
  throws InterruptedException{
    while(this.queue.size() == 0){
      wait();
    }
    if(this.queue.size() == this.limit){
      notifyAll();
    }

    return this.queue.remove(0);
  }

}

 必须注意到，在enqueue和dequeue方法内部，只有队列的大小等于上限（limit）或者下限（0）时，才调用notifyAll方法。如果队列的大小既不等于上限，也不等于下限，任何线程调用enqueue或者dequeue方法时，都不会阻塞，都能够正常的往队列中添加或者移除元素。
 
2.并发类实现
 
 
eg:使用ArrayBlockingQueue实现(摘自:http://blog.csdn.net/eson_15/article/details/51586127)
 

public class BlockingQueueTest {

    public static void main(String[] args) {
        final BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(3); //缓冲区允许放3个数据

        for(int i = 0; i < 2; i ++) {
            new Thread() { //开启两个线程不停的往缓冲区存数据

                @Override
                public void run() {
                    while(true) {
                        try {
                            Thread.sleep((long) (Math.random()*1000));
                            System.out.println(Thread.currentThread().getName() + "准备放数据"
                                    + (queue.size() == 3?"..队列已满，正在等待":"..."));
                            queue.put(1);
                            System.out.println(Thread.currentThread().getName() + "存入数据，" 
                                    + "队列目前有" + queue.size() + "个数据");
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        } 
                    }
                }

            }.start();
        }

        new Thread() { //开启一个线程不停的从缓冲区取数据

            @Override
            public void run() {
                while(true) {
                    try {
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + "准备取数据"
                                + (queue.size() == 0?"..队列已空，正在等待":"..."));
                        queue.take();
                        System.out.println(Thread.currentThread().getName() + "取出数据，" 
                                + "队列目前有" + queue.size() + "个数据");
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    } 
                }
            }
        }.start();
    }

}

 几种常用的阻塞队列
 
ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。
LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。
PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
DelayQueue：一个使用优先级队列实现的无界阻塞队列。
SynchronousQueue：一个不存储元素的阻塞队列。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。
 摘自:

 
 
http://ifeve.com/java-blocking-queue/
 
 
阻塞队列的适用场景:
 
生产者和消费者速度不一致造成的队列为空或者队列过大的情况。
 
 

 
阻塞队列
 
1. 什么是阻塞队列
2. 阻塞队列的代码使用
3. 生产者消费者模型
（1）应用一：解耦合
（2）应用二：削峰填谷
（3）相关代码
  
4.阻塞队列和生产者消费者模型功能的实现

 
1. 什么是阻塞队列
 
阻塞队列是一种特殊的队列，和数据结构中普通的队列一样，也遵守先进先出的原则
同时，阻塞队列是一种能保证线程安全的数据结构，并且具有以下两种特性：当队列满的时候，继续向队列中插入元素就会让队列阻塞，直到有其他线程从队列中取走元素；当队列为空的时候，继续出队列也会让队列阻塞，直到有其他线程往队列中插入元素
 
 
 
补充：线程阻塞的意思指代码此时不会被执行，即操作系统在此时不会把这个线程调度到CPU上去执行了
 
 
2. 阻塞队列的代码使用
 
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.BlockingDeque;
public class Test {
    public static void main(String[] args) throws InterruptedException {
        //不能直接newBlockingDeque，因为它是一个接口，要向上转型
        //LinkedBlockingDeque内部是基于链表方式来实现的
        BlockingDeque<String> queue=new LinkedBlockingDeque<>(10);//此处可以指定一个具体的数字，这里的的10代表队列的最大容量
        queue.put("hello");
        String elem=queue.take();
        System.out.println(elem);
        elem=queue.take();
        System.out.println(elem);
    }
}

 
注意： put方法带有阻塞功能，但是offer不具有，所以一般用put方法(能使用offer方法的原因是 BlockingDeque继承了Queue）
 
 打印结果如上所示，当打印了hello后，队列为空，代码执行到elem=queue.take();就不会继续往下执行了，此时线程进入阻塞等待状态，什么也不会打印了，直到有其他线程给队列中放入新的元素为止
 
3. 生产者消费者模型
 
生产者消费者模型是在服务器开发和后端开发中比较常用的编程手段，一般用于解耦合和削峰填谷。
 
 
 
高耦合度：两个代码模块的关联关系比较高
 高内聚：一个代码模块内各个元素彼此结合的紧密
 因此，我们一般追求高内聚低耦合，这样会加快执行效率，而使用生产者消费者模型就可以解耦合
 
 
（1）应用一：解耦合
 
我们以实际生活中的情况为例，这里有两台服务器：A服务器和B服务器，当A服务器传输数据给B时，要是直接传输的话，那么不是A向B推送数据，就是B从A中拉取数据，都是需要A和B直接交互，所以A和B存在依赖关系（A和B的耦合度比较高）。未来如果服务器需要扩展，比如加一个C服务器，让A给C传数据，那么改动就比较复杂，且会降低效率。这时我们可以加一个队列，这个队列为阻塞队列，如果A把数据写到队列里，B从中取，那么队列相当于是中转站（或者说交易场所），A相当于生产者（提供数据），B相当于消费者（接收数据），此时就构成了生产者消费者模型，这样会让代码耦合度更低，维护更方便，执行效率更高。
 
 
在计算机中，生产者充当其中一组线程，而消费者充当另一组线程，而交易场所就可以使用阻塞队列了
 
（2）应用二：削峰填谷
 
 
实际生活中
 在河道中大坝算是一个很重要的组成部分了，如果没有大坝，大家试想一下结果：当汛期来临后上游的水很大时，下游就会涌入大量的水发生水灾让庄稼被淹没；而旱期的话下游的水会很少可能会引发旱灾。若有大坝的话，汛期时大坝把多余的水存到大坝中，关闸蓄水，让上游的水按一定速率往下流，避免突然一波大雨把下游淹了，这样下游不至于出现水灾。旱期时大坝把之前储存好的水放出来，还是让让水按一定速率往下流，避免下流太缺水，这样既可以避免汛期发生洪涝又可以避免旱期发生旱灾了。
 峰：相当于汛期
 谷：相当于旱期
 计算机中
 这样的情况在计算机中也是很典型的，尤其是在服务器开发中，网关通常会把互联网中的请求转发给业务服务器，比如一些商品服务器，用户服务器，商家服务器（存放商家的信息），直播服务器。但因为互联网过来的请求数量是多是少不可控，相当于上游的水，如果突然来了一大波请求，网关即使能扛得住，后续的很多服务器收到很多请求也就会崩溃（处理一个请求涉及到一系列的数据库操作，因为数据库相关操作效率本身比较低，这样请求多了就处理不过来了，因此就会崩溃）
 
所以实际情况中网关和业务服务器之间往往用一个队列来缓冲，这个队列就是阻塞队列（交易场所），用这个队列来实现生产者（网关）消费者（业务服务器）模型，把请求缓存到队列中，后面的消费者（业务服务器）按照自己固定的速率去读请求。这样当请求很多时，虽然队列服务器可能会稍微受到一定压力，但能保证业务服务器的安全。
 
（3）相关代码
 
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class TestDemo {
    public static void main(String[] args) {
        // 使用一个 BlockingQueue 作为交易场所
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
        // 此线程作为消费者
        Thread customer = new Thread() {
            @Override
            public void run() {
                while (true) {
                    // 取队首元素
                    try {
                        Integer value = queue.take();
                        System.out.println("消费元素: " + value);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        customer.start();
        // 此线程作为生产者
        Thread producer = new Thread() {
            @Override
            public void run() {
                for (int i = 1; i <= 10000; i++) {
                    System.out.println("生产了元素: " + i);
                    try {
                        queue.put(i);
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        producer.start();
        try {
            customer.join();
            producer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

 

 打印如上（此代码是让生产者通过sleep每过1秒生产一个元素，而消费者不使用sleep，所以每当生产一个元素时，消费者都会立马消费一个元素）
 
4.阻塞队列和生产者消费者模型功能的实现
 
在学会如何使用BlockingQueue后，那么如何自己去实现一个呢？
 主要思路：
 
 
 
1.利用数组
 2.head代表队头，tail代表队尾
 3.head和tail重合后到底是空的还是满的判断方法：专门定义一个size记录当前队列元素个数，入队列时size加1出队列时size减1，当size为0表示空，为数组最大长度就是满的（也可以浪费一个数组空间用head和tail重合表示空，用tail+1和head重合表示满，但此方法较为麻烦，上一个方法较为直观，因此我们使用上一个方法）
 
 
public class Test2 {
    static class BlockingQueue {
    private int[] items = new int[1000];    // 此处的1000相当于队列的最大容量, 此处暂时不考虑扩容的问题.
    private int head = 0;//定义队头
    private int tail = 0;//定义队尾
    private int size = 0;//数组大小
    private Object locker = new Object();

    // put 用来入队列
    public void put(int item) throws InterruptedException {
        synchronized (locker) {
            while (size == items.length) {
                // 队列已经满了，阻塞队列开始阻塞
                locker.wait();
            }
            items[tail] = item;
            tail++;
            // 如果到达末尾, 就回到起始位置.
            if (tail >= items.length) {
                tail = 0;
            }
            size++;
            locker.notify();
        }
    }
    // take 用来出队列
    public int take() throws InterruptedException {
        int ret = 0;
        synchronized (locker) {
            while (size == 0) {
                // 对于阻塞队列来说, 如果队列为空, 再尝试取元素, 就要阻塞
                locker.wait();
            }
            ret = items[head];
            head++;
            if (head >= items.length) {
                head = 0;
            }
            size--;
            // 此处的notify 用来唤醒 put 中的 wait
            locker.notify();
        }
        return ret;
    }
}

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue queue = new BlockingQueue();
        // 消费者线程
        Thread consumer = new Thread() {
            @Override
            public void run() {
                while (true) {
                    try {
                        int elem = queue.take();
                        System.out.println("消费元素: " + elem);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        consumer.start();

        // 生产者线程
        Thread producer = new Thread() {
            @Override
            public void run() {
                for (int i = 1; i < 10000; i++) {
                    System.out.println("生产元素: " + i);
                    try {
                        queue.put(i);
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        producer.start();
        consumer.join();
        producer.join();
    }
}
 

 运行结果如上。
 注意：
 1.wait和notify的正确使用
 2.put和take都会产生阻塞情况，但阻塞条件是对立的，wait不会同时触发（put唤醒take阻塞，take唤醒put阻塞）

队列可分为两种，一种是阻塞队列，一种是非阻塞队列。
 
阻塞队列和非阻塞队列的区别：阻塞队列可以阻塞，非阻塞队列不能阻塞，只能使用队列wait(),notify()进行队列消息传送。而阻塞队列当队列里面没有值时，会阻塞直到有值输入。输入也一样，当队列满的时候，会阻塞，直到队列不为空。
 
自从Java 1.5之后，在java.util.concurrent包下提供了若干个阻塞队列，主要有以下几个：
 
ArrayBlockingQueue：
 基于数组实现的一个阻塞队列，在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性，默认情况下为非公平的，即不保证等待时间最长的队列最优先能够访问队列
 
LinkedBlockingQueue：
 基于链表实现的一个阻塞队列，在创建LinkedBlockingQueue对象时如果不指定容量大小，则默认大小为Integer.MAX_VALUE。
 
PriorityBlockingQueue：
 它会按照元素的优先级对元素进行排序，按照优先级顺序出队，每次出队的元素都是优先级最高的元素。注意，此阻塞队列为无界阻塞队列，即容量没有上限（通过源码就可以知道，它没有容器满的信号标志），前面2种都是有界队列
 
DelayQueue：
 一种延时阻塞队列，DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQue

import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用wait和notify实现Queue
 * BlockingQueue: 顾名思义，首先它是一个队列，并且支持阻塞机制，阻塞的放入和阻塞的得到数据，
 * 我们要实现LinkedBlockingQueue下面下面两个简单的方法put和take
 * put(anObject):把把对象加到BlockingQueue里面，如果BlockQueue没有空间，则调用此方法的线程被阻断
 * 直到BlockingQueue里面没有空间在继续。
 *take():取走blockingQueue里面该在首位的对象，若blockingQueue为空，阻塞进入等待状态，直到BlockingQueue有新的数据被加入
 * 
 * wait 和notify 结合synchronized使用
 * 
 *  */
public class MyQueue {

    //1.需要装元素的集合
    private final LinkedList<Object> list = new LinkedList<Object>();

    /**
     * 2.需要一个计数器，统计加入list几个的个数
     *AtomicInteger，一个提供原子操作的Integer的类。在Java语言中
     *，++i和i++操作并不是线程安全的，在使用的时候，不可避免的会用到
     *synchronized关键字。而AtomicInteger则通过一种线程安全的加减操作接口。
     * */
    private AtomicInteger count = new AtomicInteger(0);

    //指定上限上限和下限
    private final int minSize = 0;
    private final int maxSize;
    //构造方法，构造容器最大长度
    public MyQueue(int size) {
        this.maxSize = size;
    }
    public MyQueue() {
        //最大长度默认为10
        this(10);
    }

    /**
     * 初始化一个对象用于加锁
     * */
    private final Object lock = new Object();


    //返回长度
    public int size() {
        return count.get();
    }


    /**
     * 添加一个对象，如果队列满了，则阻塞
     * */
    public void put(Object obj) {
        synchronized (lock) {
            //如果容器大小刚好等于最大长度，则阻塞
            while(size() == maxSize) {
                try {
                    lock.wait();//阻塞
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            list.add(obj);
            count.incrementAndGet();//相当于i++
            System.out.println("新加入的元素为："+obj);
            lock.notify();//通知另外一个线程去取元素
        }
    }
    /**
     * 取出一个元素，如果队列为空，则阻塞
     * */
    public Object take() {
        synchronized (lock) {
        //如果容器的大小刚好等于队列最小长度，则阻塞
            while(minSize == size()) {
                try {
                lock.wait();
                } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                }
            }
            //取出第一个元素
            Object obj = list.removeFirst();
            //计数器递减
            count.decrementAndGet();
            lock.notify();//通知另外一个线程进行添加元素
            return obj;//返回结果

        }
    }
    public static void main(String[] args) {
        final MyQueue m = new MyQueue(5);

        Thread t1 = new Thread(new Runnable() {

            @Override
            public void run() {
                m.put("a");
                m.put("b");
                m.put("c");
                m.put("d");
                m.put("e");
                m.put("f");

            }
        });
        t1.start();

        try {
            Thread.sleep(1000);
            System.out.println("当前容器的大小："+m.size());
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        Thread t2 = new Thread(new Runnable() {

                    @Override
                    public void run() {
                        Object o = m.take();
                        System.out.println("取出的元素为："+o);
                    }
                });
        t2.start();
        try {
            Thread.sleep(10);
            System.out.println(m.size());
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}