Java手書きブロックキューの使用方法

// 用于保护临界区的锁
private final ReentrantLock lock;
// 用于唤醒取数据的时候被阻塞的线程
private final Condition notEmpty;
// 用于唤醒放数据的时候被阻塞的线程
private final Condition notFull;
// 用于记录从数组当中取数据的位置 也就是队列头部的位置
private int takeIndex;
// 用于记录从数组当中放数据的位置 也就是队列尾部的位置
private int putIndex;
// 记录队列当中有多少个数据
private int count;
// 用于存放具体数据的数组
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;
  // 数组的长度肯定不能够小于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 函数可能多个线程调用 但是我们需要保证在给变量赋值的时候只能够有一个线程
  // 因为如果多个线程同时进行赋值的话 那么可能后一个线程的赋值操作覆盖了前一个线程的赋值操作
  // 因此这里需要上锁
  lock.lock();
 
  try {
    // 如果队列当中的数据个数等于数组的长度的话 说明数组已经满了
    // 这个时候需要将线程挂起
    while (count == items.length)
      notFull.await(); // 将调用 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(); // 唤醒一个被 take 函数阻塞的线程唤醒
}

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

public E take() throws InterruptedException {
  // 这个函数也是不能够并发的 否则可能不同的线程取出的是同一个位置的数据
  // 进行加锁操作
  lock.lock();
  try {
    // 当 count 等于0 说明队列为空
    // 需要将线程挂起等待
    while (count == 0)
      notEmpty.await();
    // 当被唤醒之后进行出队操作
    return dequeue();
  }finally {
    lock.unlock();
  }
}
 
private E  dequeue() {
  final Object[] items = this.items;
  @SuppressWarnings("unchecked")
  E x = (E) items[takeIndex];
  items[takeIndex] = null; // 将对应的位置设置为 null GC就可以回收了
  if (++takeIndex == items.length)
    takeIndex = 0;
  count--; // 队列当中数据少一个了
  // 因为出队了一个数据 可以唤醒一个被 put 函数阻塞的线程 如果这个时候没有被阻塞的线程
  // 这个函数就不会起作用 也就说在这个函数调用之后被 put 函数挂起的线程也不会被唤醒
  notFull.signal(); // 唤醒一个被 put 函数阻塞的线程
  return x;
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
 
public class MyArrayBlockingQueue<E> {
 
  // 用于保护临界区的锁
  private final ReentrantLock lock;
  // 用于唤醒取数据的时候被阻塞的线程
  private final Condition notEmpty;
  // 用于唤醒放数据的时候被阻塞的线程
  private final Condition notFull;
  // 用于记录从数组当中取数据的位置 也就是队列头部的位置
  private int takeIndex;
  // 用于记录从数组当中放数据的位置 也就是队列尾部的位置
  private int putIndex;
  // 记录队列当中有多少个数据
  private int count;
  // 用于存放具体数据的数组
  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(&#39;]&#39;);
      }
    }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() + " 往队列当中加入数据：" + i);
        queue.put(i);
      }
    }, "生产者");
 
 
    Thread thread1 = new Thread(() -> {
      for (int i = 0; i < 10; i++) {
        try {
          System.out.println(Thread.currentThread().getName() + " 从队列当中取出数据：" + queue.take());
          System.out.println(Thread.currentThread().getName() + " 当前队列当中的数据：" + queue);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }, "消费者");
    thread.start();
    TimeUnit.SECONDS.sleep(3);
    thread1.start();
 
  }
}