一，阻塞队列使用案例
二,J.U.C 中的阻塞队列
- 1.JUC中提供的阻塞队列
- 2.阻塞队列的操作方法
  - 2.1插入操作
  - 2.2移除操作
三，ArrayBlockingQueue源码

把多线程环境比作是分布式的话，那么线程与线程之间是不是也可以使用这种消息队列的方式进行数据通信和解耦呢？

一，阻塞队列使用案例

1.注册成功后增加积分

假如模拟一个场景，就是用户注册的时候，在注册成功以后发放积分。这个场景在一般来说，会这么去实现：

但是实际上，我们需要考虑两个问题：

性能，在注册这个环节里面，假如添加用户需要花费 1 秒钟，增加积分需要花费 1 秒钟，那么整个注册结果的返回就可能需要大于 2 秒，虽然影响不是很大，但是在量比较大的时候，我们也需要做一些优化。
耦合，添加用户和增加积分，可以认为是两个领域，也就是说，增加积分并不是注册必须要具备的功能，但是一旦增加积分这个逻辑出现异常，就会导致注册失败。这种耦合在程序设计的时候是一定要规避的。

因此我们可以通过异步的方式来实现。

2.改造之前的代码逻辑

public class UserService {
    public boolean register() {
        User user = new User();
        user.setName("Mic");
        addUser(user);
        sendPoints(user);
        return true;
    }
    public static void main(String[] args) {
        new UserService().register();
    }
    private void addUser(User user) {
        System.out.println(" 添加用户：" + user);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    private void sendPoints(User user) {
        System.out.println("  发 送 积 分 给 指 定 用 户:" + user);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
@Data
class User {
    private String name;
}

3.改造之后的逻辑

public class UserService {
    private final ExecutorService single = Executors.newSingleThreadExecutor();
    private volatile boolean isRunning = true;
    ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(10);
    {
        init();
    }
    public void init() {
        single.execute(() -> {
            while (isRunning) {
                try {
                    User user = (User) arrayBlockingQueue.take();// 阻塞的方式获取队列中的数据
                    sendPoints(user);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    public boolean register() {
        User user = new User();
        user.setName("Mic");
        addUser(user);
        arrayBlockingQueue.add(user);// 添加到异步队列
        return true;
    }
    public static void main(String[] args) {
        new UserService().register();
    }
    private void addUser(User user) {
        System.out.println(" 添加用户：" + user);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    private void sendPoints(User user) {
        System.out.println("  发 送 积 分 给 指 定 用 户:" + user);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
@Data
class User {
    private String name;
}

优化以后，整个流程就变成了这样

我们使用了 ArrayBlockingQueue 基于数组的阻塞队列，来优化代码的执行逻辑。

4.阻塞队列的应用场景

阻塞队列这块的应用场景，比较多的仍然是对于生产者消费者场景的应用，但是由于分布式架构的普及，更多的关注在分布式消息队列上。所以其实如果把阻塞队列比作成分布式消息队列的话，那么所谓的生产者和消费者其实就是基于阻塞队列的解耦。

另外，阻塞队列是一个 fifo 的队列，所以对于希望在线程级别需要实现对目标服务的顺序访问的场景中，也可以使用。

二,J.U.C 中的阻塞队列

1.JUC中提供的阻塞队列

在 Java8 中，提供了 7 个阻塞队列。

ArrayBlockingQueue	数组实现的有界阻塞队列, 此队列按照先进先出（FIFO）的原则对元素进行排序。
LinkedBlockingQueue	链表实现的有界阻塞队列, 此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序
PriorityBlockingQueue	支持优先级排序的无界阻塞队列, 默认情况下元素采取自然顺序升序排列。也可以自定义类实现 compareTo()方法来指定元素排序规则，或者初始化 PriorityBlockingQueue 时，指定构造参数 Comparator 来对元素进行排序。
DelayQueue	优先级队列实现的无界阻塞队列
SynchronousQueue	不存储元素的阻塞队列, 每一个 put 操作必须等待一个 take 操作，否则不能继续添加元素。
LinkedTransferQueue	链表实现的无界阻塞队列
LinkedBlockingDeque	链表实现的双向阻塞队列

2.阻塞队列的操作方法

在阻塞队列中，提供了四种处理方式

2.1插入操作

add(e) ：添加元素到队列中，如果队列满了，继续插入元素会报错，IllegalStateException。

offer(e) : 添加元素到队列，同时会返回元素是否插入成功的状态，如果成功则返回 true。

put(e) ：当阻塞队列满了以后，生产者继续通过 put添加元素，队列会一直阻塞生产者线程，直到队列可用。

offer(e,time,unit) ：当阻塞队列满了以后继续添加元素，生产者线程会被阻塞指定时间，如果超时，则线程直接退出。

2.2移除操作

remove()：当队列为空时，调用 remove 会返回 false，如果元素移除成功，则返回 true。

poll(): 当队列中存在元素，则从队列中取出一个元素，如果队列为空，则直接返回 null。

take()：基于阻塞的方式获取队列中的元素，如果队列为空，则 take 方法会一直阻塞，直到队列中有新的数据可以消费。

poll(time,unit)：带超时机制的获取数据，如果队列为空，则会等待指定的时间再去获取元素返回。

三，ArrayBlockingQueue源码

1.构造方法

ArrayBlockingQueue 提供了三个构造方法，分别如下。
capacity：表示数组的长度，也就是队列的长度。
fair：表示是否为公平的阻塞队列，默认情况下构造的是非公平的阻塞队列。

public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);//重入锁，出队和入队持有这一把锁
    notEmpty = lock.newCondition();//初始化非空等待队列
    notFull =  lock.newCondition();//初始化非满等待队列
}
public ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);
    final ReentrantLock lock = this.lock;
    lock.lock(); // Lock only for visibility, not mutual exclusion
    try {
        int i = 0;
        try {
            for (E e : c) {
                checkNotNull(e);
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;
    } finally {
        lock.unlock();
    }
}

items 构造以后，大概是一个这样的数组结构：

2.add

以 add 方法作为入口，在 add 方法中会调用父类的 add 方法，也就是 AbstractQueue.

public boolean add(E e) {
    return super.add(e);
}
======================================================
public boolean add(E e) {
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

从父类的 add 方法可以看到，这里做了一个队列是否满了的判断，如果队列满了直接抛出一个异常。

3.offer

public boolean offer(E e) {
    //校验放入队列的元素如果为null，抛出空指针异常
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //如果队列已经满了，返回false
        if (count == items.length)
            return false;
        else {
            //否则，执行入队逻辑
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

4.checkNotNull

private static void checkNotNull(Object v) {
    if (v == null)
        throw new NullPointerException();
}

5.enqueue

private void enqueue(E x) {
    //当前队列的引用
    final Object[] items = this.items;
    //元素入队
    items[putIndex] = x;
    //如果下一个元素存放位置大于队列长度，把队列长度置为0
    if (++putIndex == items.length)
        putIndex = 0;
    //元素个数+1
    count++;
    //唤醒处于等待状态下的线程，表示当前队列中的元素不为空,如果存在消费者线程阻塞，就可以开始取出元素
    notEmpty.signal();
}

putIndex 为什么会在等于数组长度的时候重新设置为 0？

因为 ArrayBlockingQueue 是一个 FIFO 的队列，队列添加元素时，是从队尾获取 putIndex 来存储元素，当 putIndex等于数组长度时，下次就需要从数组头部开始添加了。

下面这个图模拟了添加到不同长度的元素时，putIndex 的变化，当 putIndex 等于数组长度时，不可能让 putIndex 继续累加，否则会超出数组初始化的容量大小。

当元素满了以后是无法继续添加的，因为会报错。
队列中的元素肯定会有一个消费者线程通过 take或者其他方法来获取数据，而获取数据的同时元素也会从队列中移除。

6.put

put 方法和 add 方法功能一样，差异是 put 方法如果队列满了，会阻塞。

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        //元素个数如果等于数组的长度，阻塞当前线程
        while (count == items.length)
            notFull.await();
        //否则，入队逻辑
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

7.take

take 方法是一种阻塞获取队列中元素的方法。

它的实现原理很简单，有就删除没有就阻塞，注意这个阻塞是可以中断的，如果队列没有数据那么就加入 notEmpty条件队列等待(有数据就直接取走，方法结束)，如果有新的put 线程添加了数据，那么 put 操作将会唤醒 take 线程，执行 take 操作。

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        //队列元素个数==0 阻塞
        while (count == 0)
            notEmpty.await();
        //否则执行出队逻辑
        return dequeue();
    } finally {
        lock.unlock();
    }
}

如果队列中添加了元素，那么这个时候，会在 enqueue 中调用 notempty.signal 唤醒 take 线程来获得元素

8.dequeue

这个是出队列的方法，主要是删除队列头部的元素并发返回给客户端。

private E dequeue() {
    //获取队列引用
    final Object[] items = this.items;
    //拿出来一个元素
    E x = (E) items[takeIndex];
    //将拿出的元素的索引位置设置为null
    items[takeIndex] = null;
    //如果到了数组长度  从零开始
    if (++takeIndex == items.length)
        takeIndex = 0;
    //队列元素个数-1
    count--;
    if (itrs != null)
        itrs.elementDequeued();//更新迭代器中的元素个数
    //触发 因为队列满了以后导致的被阻塞的线程
    notFull.signal();
    return x;
}

9.elementDequeued

ArrayBlockingQueue 中，实现了迭代器的功能，也就是可以通过迭代器来遍历阻塞队列中的元素。

所以 itrs.elementDequeued() 是用来更新迭代器中的元素数据的。

takeIndex 的索引变化图如下，同时随着数据的移除，会唤醒处于 put 阻塞状态下的线程来继续添加数据。

10.remove

remove 方法是移除一个指定元素。看看它的实现代码：

public boolean remove(Object o) {
    //判空
    if (o == null) return false;
    //获取队列的引用
    final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //如果队列中元素个数大于0
        if (count > 0) {
            //获取上一次放入的元素的索引位置
            final int putIndex = this.putIndex;
            //获取上一次取出的元素的索引的位置
            int i = takeIndex;
            do {
                //遍历每一个元素，如果找到了，就移除元素
                if (o.equals(items[i])) {
                    removeAt(i);
                    return true;
                }
                //这个的逻辑有啥用？
                //可能存在这样一种情况：
                // 1 null 2 3 4
                if (++i == items.length)
                    i = 0;
            } while (i != putIndex);
        }
        return false;
    } finally {
        lock.unlock();
    }
}

11.removeAt

void removeAt(final int removeIndex) {
    //获取队列的引用
    final Object[] items = this.items;
    //如果要删除的节点恰好是下一个获取元素要拿的索引位置，直接干掉就中
    if (removeIndex == takeIndex) {
        // removing front item; just advance
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
    } else {
        //这个时候就需要轮训删除
        final int putIndex = this.putIndex;
        //自旋
        for (int i = removeIndex;;) {
            //删除索引的下一个索引
            int next = i + 1;
            //如果到头了  ，意思就是  ， 那就得 从头再来
            if (next == items.length)
                next = 0;
            // 断流了 得跳过去
            if (next != putIndex) {
                items[i] = items[next];
                i = next;
            } else {
                //找到了  删除  退出 
                items[i] = null;
                this.putIndex = i;
                break;
            }
        }
        //元素个数--
        count--;
        //处理迭代器的逻辑
        if (itrs != null)
            itrs.removedAt(removeIndex);
    }
    //将往队列放元素，但是因为队列满了阻塞的线程唤醒一个，当然了，也可能队列没有线程
    notFull.signal();
}