多线程与高并发——并发编程（4）

四、阻塞队列

1 基础概念

1.1 生产者消费者概念

生产者-消费者是设计模式的一种，让生产者和消费者基于一个容器来解决强耦合的问题。生产者与消费者彼此之间不会直接通讯，而是通过一个容器（队列）进行通讯。

• 生产者生产完数据后扔到容器中，不用等消费者来处理；
• 消费者也不需要去找生产者要数据，直接从容器中获取即可；
• 而这种容器最常用的结构就是队列。

1.2 JUC阻塞队列的存取方法

常用的存取方法都来自 JUC 包下的 BlockingQueue

• 生产者存储方法：
  • add(E)：添加数据到队列，若队列满了，抛出异常；
  • offer(E)：添加数据到队列，若队列满了，返回 false；
  • offer(E,timeout,unit)：添加数据到队列，若队列满了，阻塞 timeout 时间，超时后返回 false；
  • put(E)：添加数据到队列，若队列满了，挂起线程，等到队列中有位置，再扔数据进去，死等。
• 消费者取数据方法：
  • remove()：从队列中移除数据，若队列为空，抛出异常；
  • poll()：从队列中移除数据，若队列为空，返回 false；
  • poll(timeout,unit)：从队列中移除数据，若队列为空，阻塞 timeout 时间，等生产者仍数据再获取数据，超时后返回 false；
  • take()：从队列中移除数据，若队列为空，挂起线程，一直等生产者仍数据再获取。

2 ArrayBlockingQueue

2.1 ArrayBlockingQueue的基本使用

• ArrayBlockingQueue 在初始化时，必须指定当前队列的长度，因为 ArrayBlockingQueue 是基于数组实现的队列结构，数组长度不可变，必须提前设置数据长度信息。

public static void main(String[] args) throws InterruptedException {// 必须设置队列长度ArrayBlockingQueue queue = new ArrayBlockingQueue(4);// 生产者生产数据queue.add("1");queue.offer("2");queue.offer("3", 2, TimeUnit.SECONDS);queue.put("4");// 消费者消费数据System.out.println(queue.remove());System.out.println(queue.poll());System.out.println(queue.poll(2, TimeUnit.SECONDS));System.out.println(queue.take());
}

2.2 生产者方法实现原理

• 生产者添加数据到队列的方法比较多，需要一个一个看

2.2.1 ArrayBlockingQueue的常见属性

ArrayBlockingQueue中的成员变量

final Object[] items; 				// 就是数组本身
int takeIndex;						// 取数据的下标
int putIndex;						// 存数据的下标
int count;							// 当前数组中元素的个数
final ReentrantLock lock;			// 就是一个 ReentrantLock 锁
private final Condition notEmpty;	// 消费者挂起线程和唤醒线程用到的Condition（可看作是synchronized的wait和notify)
private final Condition notFull;	// 生产者挂起线程和唤醒线程用到的Condition（可看作是synchronized的wait和notify)

2.2.2 add方法

• add方法本身就是调用了offer方法，如果offer方法返回false，直接抛出异常

public boolean add(E e) {if (offer(e))return true;else 	// 抛出的异常throw new IllegalStateException("Queue full");
}

2.2.3 offer方法

public boolean offer(E e) {checkNotNull(e);	// 要求存储的数据不允许为null，否则抛出空指针异常// 拿到当前阻塞队列的lock锁final ReentrantLock lock = this.lock;lock.lock();	// 为保证线程安全，加锁try {// 判断队列中元素是否满了，若满了，则返回falseif (count == items.length)return false;else {// 队列没满，执行 enqueue 将元素添加到队列中，并返回trueenqueue(e);return true;}} finally {lock.unlock();		// 操作完释放锁}
}
// ================
private void enqueue(E x) {// 拿到数组的引用，将元素放到指定的位置final Object[] items = this.items;items[putIndex] = x;// 对putIndex进行++操作，并判断是否等于数组长度，需要归为if (++putIndex == items.length)putIndex = 0;	// 归位：将索引值设置为0count++;	// 添加成功，数据++notEmpty.signal();	// 将一个Condition中阻塞的线程唤醒
}

2.2.4 offer(time,unit)方法

生产者在添加数据时，如果队列已经满，阻塞一会：

• 阻塞到消费者消费了消息，然后唤醒当前阻塞线程；
• 阻塞到了 timeout 时间，再次判断是否可以添加，若不能直接告辞。

// 线程在挂起时，如果对当前阻塞线程的终端标记位进行设置，会抛出异常直接结束
public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {// 非空校验checkNotNull(e);long nanos = unit.toNanos(timeout);		// 将时间单位转为纳秒final ReentrantLock lock = this.lock;	// 加锁lock.lockInterruptibly();	// 允许线程中断排除异常的加锁方法try {// 为什么是while（虚假唤醒）while (count == items.length) {	// 如果元素个数和数组长度一致，说明队列满了if (nanos <= 0)	// 判断等待时间是否充裕return false;	// 不充裕，直接添加失败，返回false// 挂起等待，会同时释放锁资源（对标 synchronized 的wait方法）// awaitNanos会挂起线程，并且返回剩余的阻塞时间，恢复执行时，需要重新获取锁资源nanos = notFull.awaitNanos(nanos);}enqueue(e); // 这里锁门队列有空间了，enqueue将数据添加到阻塞队列中，并返回truereturn true;} finally {lock.unlock();	// 是否锁资源}
}

2.2.5 put方法

• 如果队列是满的，就一直挂起，直到被唤醒，或者被中断

public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == items.length)// await方法会一直阻塞，直到被唤醒或者被中断notFull.await();enqueue(e);} finally {lock.unlock();}
}

2.3 消费者方法实现原理

2.3.1 remove方法

• remove方法本身就是调用了poll方法，如果poll方法返回null，直接抛出异常

public E remove() {E x = poll();if (x != null)return x;else	// 没数据抛出异常throw new NoSuchElementException();
}

2.3.2 poll方法

// 拉取数据
public E poll() {final ReentrantLock lock = this.lock;lock.lock();	// 加锁try {// 若没有数据，直接返回null；否则执行dequeue，取出数据并返回return (count == 0) ? null : dequeue();} finally {lock.unlock();}
}
// 取出数据
private E dequeue() {// 将成员变量引用到局部变量final Object[] items = this.items;@SuppressWarnings("unchecked")E x = (E) items[takeIndex];		// 直接获取指定索引位置的数据items[takeIndex] = null;		// 取出数据后，清空该索引位置if (++takeIndex == items.length)	// 设置下次取数据的索引位置takeIndex = 0;count--;	// 数组中元素个数减一if (itrs != null)	// 迭代器内容先跳过itrs.elementDequeued();// signal方法，会唤醒当前Condition中排队的一个Node// signalAll方法，会将Condition中所有的Node，全都唤醒notFull.signal();return x;	// 返回数据
}

2.3.3 poll(timeout,unit)方法

public E poll(long timeout, TimeUnit unit) throws InterruptedException {long nanos = unit.toNanos(timeout);		// 转换时间单位final ReentrantLock lock = this.lock;lock.lockInterruptibly();				// 加锁，可中断唤醒try {while (count == 0) {	// 如果没数据if (nanos <= 0)		// 也没时间了，就不阻塞，返回nullreturn null;// 有时间，就挂起消费者线程一段时间nanos = notEmpty.awaitNanos(nanos);}return dequeue();	// 取数据} finally {lock.unlock();}
}

2.3.4 take方法

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)	// 使用while，防止虚假唤醒notEmpty.await();return dequeue();} finally {lock.unlock();}
}

2.3.5 虚假唤醒

阻塞队列中，如果需要线程挂起操作，判断有无数据的位置采用的是while循环，为什么不使用if？

• 首先肯定不能换成 if 逻辑判断，比如：有线程 A、B、E、C，其中 ABE 是生产者，C是消费者。假如线程的队列是满的，AB挂起

// E，拿到锁资源，还没有走while判断
while (count == items.length)// A醒了// B挂起notFull.await();
enqueue(e)；

• C 此时消费一条数据，执行 notFull.signal() 唤醒一个线程，A线程被唤醒；E走判断发现有空余位置，可以添加数据到队列，则E添加数据，走enqueue。
• 如果判断是 if，A 在E释放锁资源后，拿到锁资源，直接走 enqueue 方法，此时 A线程就是在 putIndex 的位置，覆盖掉之前的数据，会造成数据安全问题。

3 LinkedBlockingQueue

3.1 LinkedBlockingQueue的底层实现

• 查看 LinkedBlockingQueue 是如何存储数据，以及如何实现链表结构的。

// Node对象就是存储数据的单位
static class Node<E> {// 存储的数据E item;// 指向下一个数据的指针Node<E> next;// 有参构造Node(E x) { item = x; }
}

• 查看LinkedBlockingQueue的有参构造

// 可以手动指定LinkedBlockingQueue的长度，如果没有指定，默认为Integer.MAX_VALUE
public LinkedBlockingQueue(int capacity) {if (capacity <= 0) throw new IllegalArgumentException();this.capacity = capacity;// 在初始化时，构建一个item为null的节点，作为head和last，这种node可以成为哨兵Node，// 如果没有哨兵节点，那么在获取数据时，需要判断head是否为null，才能找next// 如果没有哨兵节点，那么在添加数据时，需要判断last是否为null，才能找nextlast = head = new Node<E>(null);
}

• 查看LinkedBlockingQueue的其他属性

// 因为是链表，没有想数组的length属性，基于AtomicInteger来记录长度
private final AtomicInteger count = new AtomicInteger();
transient Node<E> head;	// 链表的头，用于取数据
private transient Node<E> last;	// 链表的尾，用于存数据
// 消费者的锁
private final ReentrantLock takeLock