详解Java如何实现一个BlockingQueue

public class BlockingQueueWithLoop implements Queue {
​
  private int[] elements;
  private int head;
  private int tail;
  private volatile int size; // 队列元素个数
​
  public BlockingQueueWithLoop() {
    this(10);
  }
​
  public BlockingQueueWithLoop(int capacity) {
    this.elements = new int[capacity];
    this.head = 0;
    this.tail = 0;
    this.size = 0;
  }
​
  @Override
  public void put(int e) throws InterruptedException {
    while (size == elements.length) {} // 使用自旋锁，等待队列不满
    synchronized (this) {
      if(size == elements.length){ // 双重检索
        return;
      }
      elements[tail] = e;
      tail++;
      if (tail == elements.length) {
        tail = 0;
      }
      size++;
    }
  }
​
  @Override
  public int take() throws InterruptedException {
    while (true) {
      while (size <= 0) {} // 使用自旋锁，等待队列不为空
​
      synchronized (this) { // 队列不为空，需要加锁
        if (size > 0) {     // 双重检索
          int e = elements[head];
          head++;
          if (head == elements.length) {
            head = 0;
          }
          size--;
          return e;
        }
      }
    }
  }
​
  @Override
  public synchronized int size() {
    return size;
  }
​
  @Override
  public synchronized boolean isEmpty() {
    return size == 0;
  }
}

public class Object {
  public final void wait() throws InterruptedException;
  public final native void wait(long timeoutMillis) throws InterruptedException;
  public final void wait(long timeoutMillis, int nanos) throws InterruptedException 
  public final native void notify();
  public final native void notifyAll();
}

public class BlockingQueueWithSync implements Queue {
​
  private int[] elements;
  private int head;
  private int tail;
  private volatile int size; // 队列元素个数
​
  public BlockingQueueWithSync() {
    this(10);
  }
​
  public BlockingQueueWithSync(int capacity) {
    this.elements = new int[capacity];
    this.head = 0;
    this.tail = 0;
    this.size = 0;
  }
​
  @Override
  public synchronized void put(int e) throws InterruptedException {
    // 当队列满的时候阻塞
    while (size == elements.length) {
      this.wait();
    }
    elements[tail] = e;
    tail++;
    if (tail == elements.length) {
      tail = 0;
    }
    size++;
    // 通知其他线程有数据了
    this.notifyAll();
  }
​
  @Override
  public synchronized int take() throws InterruptedException {
    // 当队列空的时候阻塞
    while (isEmpty()) {
      this.wait();
    }
    int e = elements[head];
    if (++head == elements.length) {
      head = 0;
    }
    --size;
    // 通知其他线程，暂无数据
    this.notifyAll();
    return e;
  }
​
  @Override
  public synchronized int size() {
    return size;
  }
​
  @Override
  public synchronized boolean isEmpty() {
    return size == 0;
  }
}

// ObjectMonitor.cpp
void ObjectMonitor::wait(jlong millis, bool interruptible, TRAPS) {}
void ObjectMonitor::notify(TRAPS) {}
void ObjectMonitor::notifyAll(TRAPS) {}

public class BlockingQueueWithCondition implements Queue {
​
  private int[] elements;
  private int head;
  private int tail;
  private volatile int size; // 队列元素个数
  private final ReentrantLock lock = new ReentrantLock();
  private final Condition notEmpty = lock.newCondition();
  private final Condition notFull = lock.newCondition();
​
  public BlockingQueueWithCondition() {
    this(10);
  }
​
  public BlockingQueueWithCondition(int capacity) {
    this.elements = new int[capacity];
    this.head = 0;
    this.tail = 0;
    this.size = 0;
  }
​
  @Override
  public void put(int e) throws InterruptedException {
    lock.lockInterruptibly();
    try {
      while (size == elements.length) {
        notFull.await();
      }
      elements[tail] = e;
      tail++;
      if (tail == elements.length) {
        tail = 0;
      }
      size++;
      notEmpty.signalAll();
    } finally {
      lock.unlock();
    }
  }
​
  @Override
  public int take() throws InterruptedException {
    lock.lockInterruptibly();
    try {
      while (isEmpty()) {
        notEmpty.await();
      }
      int e = elements[head];
      if (++head == elements.length) {
        head = 0;
      }
      --size;
      notFull.signalAll();
      return e;
    } finally {
      lock.unlock();
    }
  }
​
  @Override
  public int size() {
    try {
      lock.lock();
      return size;
    } finally {
      lock.unlock();
    }
  }
​
  @Override
  public boolean isEmpty() {
    try {
      lock.lock();
      return size == 0;
    } finally {
      lock.unlock();
    }
  }
}

// java.util.concurrent.locks.Condition
public interface Condition {
    void await() throws InterruptedException;
    void awaitUninterruptibly();
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    boolean awaitUntil(Date deadline) throws InterruptedException;
    void signal();
    void signalAll();
}

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    /** First node of condition queue. */
    private transient Node firstWaiter;
    /** Last node of condition queue. */
    private transient Node lastWaiter;
    // ...
}
​
static final class Node {
  volatile int waitStatus;
  volatile Node prev;
  volatile Node next;
  volatile Thread thread;
  Node nextWaiter; // 用于Condition
}

public final void await() throws InterruptedException {
    // 检测到中断，抛异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 将线程包裹为Node添加到Condition等待队列尾部
    Node node = addConditionWaiter();
    // 将state修改为0，返回释放前锁的状态
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) { // 被意外唤醒的话需要再次挂起
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 接收到 signal，返回前需要再排队等待锁
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}
​
// 加入条件等待队列
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
​
    // 加入链表末尾
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}
​
​
// 意外唤醒
final boolean isOnSyncQueue(Node node) {
    // 进入同步队列时，waitStatus为0,且prev指向前驱节点
    // 之后节点可能被取消，状态变为CANCELLED
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    // 存在后继节点，肯定在同步队列中
    if (node.next != null) 
        return true;
    // 兜底，从tail查找，确保node已经被加入同步队列
    return findNodeFromTail(node);
}
​

public final void signal() {
  // 必须保证持有锁
  if (!isHeldExclusively())
    throw new IllegalMonitorStateException();
  Node first = firstWaiter;
  if (first != null)
    // 唤醒队首线程
    doSignal(first);
}
​
private void doSignal(Node first) {
  do {
    // 将first移出队列
    if ( (firstWaiter = first.nextWaiter) == null)
      lastWaiter = null;
    first.nextWaiter = null;
  } while (!transferForSignal(first) &&  // 唤醒线程
           (first = firstWaiter) != null);
}
​
final boolean transferForSignal(Node node) {
  // 节点状态不为CONDITION，说明已经被取消了，不进行唤醒
  if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
    return false;
  // 将节点加入到同步队列，返回之前的队尾节点
  Node p = enq(node);
  int ws = p.waitStatus;
  // 如果设置前驱节点的状态失败（如前驱已被取消）则直接唤醒线程
  // 唤醒后的线程会在 `await` 中执行 `acquireQueued` 直到抢锁成功
  if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
    LockSupport.unpark(node.thread);
  return true;
}