JUC之阻塞队列BlockingQueue的实现原理

1. 阻塞队列首先它是一个队列，是队列就会遵循先进先出（FIFO）的原则，又因为它是阻塞的，故与普通的队列有两点区别：

　 A. 当一个线程向队列里面添加数据时，如果队列是满的，那么将阻塞该线程，暂停添加数据。

　 B. 当一个线程从队列里面取出数据时，如果队列是空的，那么将阻塞该线程，暂停取出数据。

2. JUC中实现一个阻塞队列一般都会实现BlockingQueue接口，主要方法说明：

方法/处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入数据方法	add(e)	offer(e)	put(e)	offer(e, timeout, unit)
移除数据方法	remove()	poll()	take()	pull(timeout, unit)
检查方法	element()	peek()	不可用	不可用

注："返回特殊值"的意思是说当向队列插入（offer）数据时，会返回数据是否插入成功，成功返回true。如果是移除方法（poll），则是从队列里面取出一个数据，如果没有就返回null。

3. jdk里提供的阻塞队列

   1). ArrayBlockingQueue

　　基于数组的阻塞队列实现，其内部维护一个定长的数组，用于存储队列元素。线程阻塞的实现是通过ReentrantLock来完成的，数据的插入与取出共用同一个锁，因此ArrayBlockingQueue并不能实现生产、消费同时进行。而且在创建ArrayBlockingQueue时，我们还可以控制对象的内部锁是否采用公平锁，默认采用非公平锁。

  2). LinkedBlockingQueue

　　基于单向链表的阻塞队列实现，在初始化LinkedBlockingQueue的时候可以指定对立的大小，也可以不指定，默认类似一个无限大小的容量（Integer.MAX_VALUE），不指队列容量大小也是会有风险的，一旦数据生产速度大于消费速度，系统内存将有可能被消耗殆尽，因此要谨慎操作。另外LinkedBlockingQueue中用于阻塞生产者、消费者的锁是两个（锁分离），因此生产与消费是可以同时进行的。

  3). PriorityBlockingQueue  一个支持优先级排序的无界阻塞队列

  4). SynchronousBlockingQueue  一个不存储原色的阻塞队列

  5). DelayQueue  一个使用优先级队列队列实现的无界阻塞队列

4. 实现一个简单的阻塞队列，基于List实现，生产、消费共用同一个锁，而且是一个有界的队列

　1). 实现阻塞队：MyBlockingQueue

package com.winter.juc;

import java.util.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class MyBlockingQueue<E> {

    private int count;
    private int capacity;
    private final List<E> containor;

    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    public MyBlockingQueue(int capacity) {
        containor = new ArrayList<E>(capacity);
        this.count = 0;
        this.capacity = capacity;
    }

    public void put(E data) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (count == capacity) {
                notFull.await();
            }
            containor.add(data);
            this.count++;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            this.count--;
            notFull.signal();
            E node = containor.get(0);
            containor.remove(0);
            return node;
        } finally {
            lock.unlock();
        }
    }
}

2).接下来使用该队列

生产者：

package com.winter.juc;

import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.RandomUtils;

public class Productor implements Runnable {
   private MyBlockingQueue<NodeItem> queue;

   public Productor(MyBlockingQueue<NodeItem> queue) {
      this.queue = queue;
   }

   public void run() {
      while (true) {
         try {
            TimeUnit.SECONDS.sleep(RandomUtils.nextInt(0, 5));
            NodeItem node = new NodeItem();
            node.setKey(GlobalKey.get());
            System.out.println("produce a node" + node);
            queue.put(node);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }
   }
}

消费者：

package com.winter.juc;

import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.RandomUtils;

public class Customer implements Runnable {
    private MyBlockingQueue<NodeItem> queue;
    public Customer(MyBlockingQueue<NodeItem> queue) {
        this.queue = queue;
    }

    public void run() {
        while (true) {
            try {
                TimeUnit.SECONDS.sleep(RandomUtils.nextInt(0, 5));
                NodeItem node = queue.take();
                System.out.println("consume a node" + node);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

元素实体：

package com.winter.juc;
public class NodeItem {

    private int key;

    public int getKey() {
        return key;
    }

    public void setKey(int key) {
        this.key = key;
    }

    @Override
    public String toString() {
        return "NodeItem{" +
                "key=" + key +
                '}';
    }
}

全局键值：

package com.winter.juc;
import java.util.concurrent.atomic.AtomicInteger;
public class GlobalKey {

    private static AtomicInteger key = new AtomicInteger(1);

    public static int get() {
        return key.getAndIncrement();
    }
}

main方法：

package com.winter.juc;

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class ProviderService {
    public static void main(String[] args) {

        final MyBlockingQueue blockingQueue = new MyBlockingQueue(3);

        ExecutorService exec = Executors.newCachedThreadPool();

        exec.submit(new Productor(blockingQueue));
        exec.submit(new Customer(blockingQueue));
    }
}

执行结果：

product a nodeNodeItem{key=1}
custom a nodeNodeItem{key=1}
product a nodeNodeItem{key=2}
custom a nodeNodeItem{key=2}
product a nodeNodeItem{key=3}
custom a nodeNodeItem{key=3}
product a nodeNodeItem{key=4}
custom a nodeNodeItem{key=4}
product a nodeNodeItem{key=5}
product a nodeNodeItem{key=6}
custom a nodeNodeItem{key=5}
product a nodeNodeItem{key=7}
custom a nodeNodeItem{key=6}
custom a nodeNodeItem{key=7}
product a nodeNodeItem{key=8}
custom a nodeNodeItem{key=8}
product a nodeNodeItem{key=9}
custom a nodeNodeItem{key=9}
product a nodeNodeItem{key=10}
product a nodeNodeItem{key=11}
product a nodeNodeItem{key=12}
custom a nodeNodeItem{key=10}
custom a nodeNodeItem{key=11}
product a nodeNodeItem{key=13}
product a nodeNodeItem{key=14}
custom a nodeNodeItem{key=12}
product a nodeNodeItem{key=15}
custom a nodeNodeItem{key=13}
product a nodeNodeItem{key=16}
product a nodeNodeItem{key=17}
custom a nodeNodeItem{key=14}
product a nodeNodeItem{key=18}
custom a nodeNodeItem{key=15}
product a nodeNodeItem{key=19}
custom a nodeNodeItem{key=16}
custom a nodeNodeItem{key=17}
custom a nodeNodeItem{key=18}
product a nodeNodeItem{key=20}
custom a nodeNodeItem{key=19}
custom a nodeNodeItem{key=20}
product a nodeNodeItem{key=21}
custom a nodeNodeItem{key=21}
product a nodeNodeItem{key=22}
custom a nodeNodeItem{key=22}

可以看到product和custom是对应执行的。

后续会继续实现一个类似LinkedBlockingQueue的阻塞队列，支持锁分离。

注：本人水平有限，如有问题，欢迎交流指出。