// 用于保護臨界區(qū)的鎖
private final ReentrantLock lock;
// 用于喚醒取數(shù)據(jù)的時候被阻塞的線程
private final Condition notEmpty;
// 用于喚醒放數(shù)據(jù)的時候被阻塞的線程
private final Condition notFull;
// 用于記錄從數(shù)組當(dāng)中取數(shù)據(jù)的位置 也就是隊列頭部的位置
private int takeIndex;
// 用于記錄從數(shù)組當(dāng)中放數(shù)據(jù)的位置 也就是隊列尾部的位置
private int putIndex;
// 記錄隊列當(dāng)中有多少個數(shù)據(jù)
private int count;
// 用于存放具體數(shù)據(jù)的數(shù)組
private Object[] items;

@SuppressWarnings("unchecked")
public MyArrayBlockingQueue(int size) {
  this.lock = new ReentrantLock();
  this.notEmpty = lock.newCondition();
  this.notFull = lock.newCondition();
  // 其實可以不用初始化 類會有默認初始化 默認初始化為0
  takeIndex = 0;
  putIndex = 0;
  count = 0;
  // 數(shù)組的長度肯定不能夠小于0
  if (size <= 0)
    throw new RuntimeException("size can not be less than 1");
  items = (E[])new Object[size];
}

public void put(E x){
  // put 函數(shù)可能多個線程調(diào)用 但是我們需要保證在給變量賦值的時候只能夠有一個線程
  // 因為如果多個線程同時進行賦值的話 那么可能后一個線程的賦值操作覆蓋了前一個線程的賦值操作
  // 因此這里需要上鎖
  lock.lock();
 
  try {
    // 如果隊列當(dāng)中的數(shù)據(jù)個數(shù)等于數(shù)組的長度的話 說明數(shù)組已經(jīng)滿了
    // 這個時候需要將線程掛起
    while (count == items.length)
      notFull.await(); // 將調(diào)用 await的線程掛起
    // 當(dāng)數(shù)組沒有滿 或者在掛起之后再次喚醒的話說明數(shù)組當(dāng)中有空間了
    // 這個時候需要將數(shù)組入隊 
    // 調(diào)用入隊函數(shù)將數(shù)據(jù)入隊
    enqueue(x);
  } catch (InterruptedException e) {
    e.printStackTrace();
  } finally {
    // 解鎖
    lock.unlock();
  }
}
 
// 將數(shù)據(jù)入隊
private void enqueue(E x) {
  this.items[putIndex] = x;
  if (++putIndex == items.length)
    putIndex = 0;
  count++;
  notEmpty.signal(); // 喚醒一個被 take 函數(shù)阻塞的線程喚醒
}

public boolean offer(E e) {
  final ReentrantLock lock = this.lock;
  lock.lock();
  try {
    // 如果數(shù)組滿了 則直接返回false 而不是被阻塞
    if (count == items.length)
      return false;
    else {
      // 如果數(shù)組沒有滿則直接入隊 并且返回 true
      enqueue(e);
      return true;
    }
  } finally {
    lock.unlock();
  }
}

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

public E take() throws InterruptedException {
  // 這個函數(shù)也是不能夠并發(fā)的 否則可能不同的線程取出的是同一個位置的數(shù)據(jù)
  // 進行加鎖操作
  lock.lock();
  try {
    // 當(dāng) count 等于0 說明隊列為空
    // 需要將線程掛起等待
    while (count == 0)
      notEmpty.await();
    // 當(dāng)被喚醒之后進行出隊操作
    return dequeue();
  }finally {
    lock.unlock();
  }
}
 
private E  dequeue() {
  final Object[] items = this.items;
  @SuppressWarnings("unchecked")
  E x = (E) items[takeIndex];
  items[takeIndex] = null; // 將對應(yīng)的位置設(shè)置為 null GC就可以回收了
  if (++takeIndex == items.length)
    takeIndex = 0;
  count--; // 隊列當(dāng)中數(shù)據(jù)少一個了
  // 因為出隊了一個數(shù)據(jù) 可以喚醒一個被 put 函數(shù)阻塞的線程 如果這個時候沒有被阻塞的線程
  // 這個函數(shù)就不會起作用 也就說在這個函數(shù)調(diào)用之后被 put 函數(shù)掛起的線程也不會被喚醒
  notFull.signal(); // 喚醒一個被 put 函數(shù)阻塞的線程
  return x;
}

@Override
public String toString() {
  StringBuilder stringBuilder = new StringBuilder();
  stringBuilder.append("[");
  // 這里需要上鎖 因為我們在打印的時候需要打印所有的數(shù)據(jù)
  // 打印所有的數(shù)據(jù)就需要對數(shù)組進行遍歷操作 而在進行遍歷
  // 操作的時候是不能進行插入和刪除操作的 因為打印的是某
  // 個時刻的數(shù)據(jù)
  lock.lock();
  try {
    if (count == 0)
      stringBuilder.append("]");
    else {
      int cur = 0;
      // 對數(shù)據(jù)進行遍歷 一共遍歷 count 次 因為數(shù)組當(dāng)中一共有 count
      // 個數(shù)據(jù)
      while (cur != count) {
        // 從 takeIndex 位置開始進行遍歷 因為數(shù)據(jù)是從這個位置開始的
        stringBuilder.append(items[(cur + takeIndex) % items.length].toString() + ", ");
        cur += 1;
      }
      // 刪除掉最后一次沒用的 ", "
      stringBuilder.delete(stringBuilder.length() - 2, stringBuilder.length());
      stringBuilder.append(']');
    }
  }finally {
    lock.unlock();
  }
  return stringBuilder.toString();
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
 
public class MyArrayBlockingQueue<E> {
 
  // 用于保護臨界區(qū)的鎖
  private final ReentrantLock lock;
  // 用于喚醒取數(shù)據(jù)的時候被阻塞的線程
  private final Condition notEmpty;
  // 用于喚醒放數(shù)據(jù)的時候被阻塞的線程
  private final Condition notFull;
  // 用于記錄從數(shù)組當(dāng)中取數(shù)據(jù)的位置 也就是隊列頭部的位置
  private int takeIndex;
  // 用于記錄從數(shù)組當(dāng)中放數(shù)據(jù)的位置 也就是隊列尾部的位置
  private int putIndex;
  // 記錄隊列當(dāng)中有多少個數(shù)據(jù)
  private int count;
  // 用于存放具體數(shù)據(jù)的數(shù)組
  private Object[] items;
 
 
  @SuppressWarnings("unchecked")
  public MyArrayBlockingQueue(int size) {
    this.lock = new ReentrantLock();
    this.notEmpty = lock.newCondition();
    this.notFull = lock.newCondition();
    // 其實可以不用初始化 類會有默認初始化 默認初始化為0
    takeIndex = 0;
    putIndex = 0;
    count = 0;
    if (size <= 0)
      throw new RuntimeException("size can not be less than 1");
    items = (E[])new Object[size];
  }
 
  public void put(E x){
    lock.lock();
 
    try {
      while (count == items.length)
        notFull.await();
      enqueue(x);
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
  }
 
  private void enqueue(E x) {
    this.items[putIndex] = x;
    if (++putIndex == items.length)
      putIndex = 0;
    count++;
    notEmpty.signal();
  }
 
  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--;
    notFull.signal();
    return x;
  }
 
  public boolean add(E e) {
    if (offer(e))
      return true;
    else
      throw new RuntimeException("Queue full");
  }
 
  public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      if (count == items.length)
        return false;
      else {
        enqueue(e);
        return true;
      }
    } finally {
      lock.unlock();
    }
  }
 
  public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      return (count == 0) ? null : dequeue();
    } finally {
      lock.unlock();
    }
  }
 
  public E take() throws InterruptedException {
    lock.lock();
    try {
      while (count == 0)
        notEmpty.await();
      return dequeue();
    }finally {
      lock.unlock();
    }
  }
 
  @Override
  public String toString() {
    StringBuilder stringBuilder = new StringBuilder();
    stringBuilder.append("[");
    lock.lock();
    try {
      if (count == 0)
        stringBuilder.append("]");
      else {
        int cur = 0;
        while (cur != count) {
          stringBuilder.append(items[(cur + takeIndex) % items.length].toString()).append(", ");
          cur += 1;
        }
        stringBuilder.delete(stringBuilder.length() - 2, stringBuilder.length());
        stringBuilder.append(']');
      }
    }finally {
      lock.unlock();
    }
    return stringBuilder.toString();
  }
 
}

import java.util.concurrent.TimeUnit;
 
public class Test {
 
  public static void main(String[] args) throws InterruptedException {
    MyArrayBlockingQueue<Integer> queue = new MyArrayBlockingQueue<>(5);
    Thread thread = new Thread(() -> {
      for (int i = 0; i < 10; i++) {
        System.out.println(Thread.currentThread().getName() + " 往隊列當(dāng)中加入數(shù)據(jù)：" + i);
        queue.put(i);
      }
    }, "生產(chǎn)者");
 
 
    Thread thread1 = new Thread(() -> {
      for (int i = 0; i < 10; i++) {
        try {
          System.out.println(Thread.currentThread().getName() + " 從隊列當(dāng)中取出數(shù)據(jù)：" + queue.take());
          System.out.println(Thread.currentThread().getName() + " 當(dāng)前隊列當(dāng)中的數(shù)據(jù)：" + queue);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }, "消費者");
    thread.start();
    TimeUnit.SECONDS.sleep(3);
    thread1.start();
 
  }
}