• java数据结构--阻塞队列

    目录

    一.概念

    二.生产者消费者问题

    三.阻塞队列接口BlockingQueue

    四.基于数组实现单锁的阻塞队列

    1.加锁方式

    2.代码实现

    3.解释说明

    (1).offer添加元素

    （2）poll取出元素

    4.timeout超时时间

    5.测试

    五.基于数组实现双锁的阻塞队列

    1.问题

    2.关于size共享变量

    3.死锁问题

     4.级联唤醒

    （1）offer中只唤醒一次，其他交给poll线程唤醒

    (2)poll中只唤醒一次，其他交给offer线程唤醒

    5.完整代码

    一.概念

    阻塞队列是一种特殊类型的队列，具有额外的阻塞操作。在阻塞队列中，当队列为空时，从队列中获取元素的操作会被阻塞，直到有元素被添加到队列中为止；当队列满时，向队列中添加元素的操作会被阻塞，直到队列有空闲位置为止。

    阻塞队列在多线程编程中非常有用，可以有效地进行线程间的协调和通信。它提供了一种线程安全的方式来共享数据，避免了常见的并发问题，如资源争用和死锁。

    常见的阻塞队列有以下几种实现方式：

    1. ArrayBlockingQueue：基于数组实现的有界阻塞队列，需要指定队列的容量。
    2. LinkedBlockingQueue：基于链表实现的可选有界或无界阻塞队列。
    3. PriorityBlockingQueue：基于堆实现的无界优先级阻塞队列，元素按照优先级进行排序。
    4. SynchronousQueue：特殊的阻塞队列，每个插入操作必须等待一个对应的删除操作，反之亦然。

    阻塞队列提供了一些常用的操作方法，如put()和take()等。put()方法用于向队列中添加元素，并在队列满时阻塞调用线程；take()方法用于获取队列中的元素，并在队列为空时阻塞调用线程。

    使用阻塞队列可以简化并发编程的实现，提高代码的可读性和维护性。它能够有效地控制线程的访问顺序，并提供了一种直观的方式来处理线程间的同步问题。

    二.生产者消费者问题

     我们先来分析一下，在我们之前学过的操作系统中，有一个生产者和消费者问题，当生产者每次生产完一个产品后，要往缓冲区中放入，但是如果此时缓冲区是满的，那么就要让生产者进入阻塞状态，进行等待，当消费者从缓冲区中取出一个产品消费后，缓冲区有了空位，此时生产者就被唤醒，往缓冲区中放入生产的产品，如果有多个生产者和消费者，此时为了防止出现混乱，就要在放入缓冲区之前加锁，放入后解锁，消费者在从缓冲区取之前也要加锁，取出后再解锁

    三.阻塞队列接口BlockingQueue

    1. /**
    2.  * 阻塞队列接口定义
    3.  * @param 
    4.  */
    5. public interface BlockingQueue {
    6.  
    7.     void offer(E e) throws InterruptedException;
    8.  
    9.     boolean offer(E e,long timeout) throws InterruptedException;
    10.  
    11.     E poll() throws InterruptedException;
    12.  
    13.     E poll(long timeout) throws InterruptedException;
    14.  
    15.  
    16.    boolean isFull();
    17.  
    18.    boolean isEmpty();
    19.  
    20. }

      我们定义了接口，其中，主要就是成对的两个方法,offer和poll,还有在此基础上加入超时时间的，加入超时时间是，当队列满时，超过了等待时间，就不去添加了，直接返回失败false

    四.基于数组实现单锁的阻塞队列

    1.加锁方式

      这里我们先说明一下，在java中，可以有两种方式加锁

      1.synocized关键字加锁

     2.ReentrantLock类创建可重入锁对象,该对象还可以创建出条件对象，来执行阻塞和唤醒操作

    2.代码实现

      我们先来用单锁实现一下阻塞队列

    1. /**
    2.  * 基于数组实现阻塞队列
    3.  * @param 
    4.  */
    5. public class ArrayBlockQueue implements BlockingQueue{
    6.  
    7.  
    8.     private final E[] array;
    9.     private int head;
    10.     private int tail;
    11.     private int size;
    12.  
    13.     @SuppressWarnings("all")
    14.     public ArrayBlockQueue(int capacity){
    15.         array = (E[]) new Object[capacity];
    16.     }
    17.  
    18.     //锁对象
    19.     private final ReentrantLock reentrantLock = new ReentrantLock();
    20.     //条件对象
    21.     private final Condition headWaits = reentrantLock.newCondition();
    22.     private final Condition tailWaits = reentrantLock.newCondition();
    23.  
    24.     /**
    25.      * 向队尾添加元素，如果队列为满，则阻塞当前线程
    26.      * @param e 要添加元素
    27.      * @throws InterruptedException
    28.      */
    29.     @Override
    30.     public void offer(E e) throws InterruptedException {
    31.        //先加锁
    32.         reentrantLock.lock();
    33.         try{
    34.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    35.             while (isFull()){
    36.                 //如果是满的，就让当前线程阻塞
    37.                 tailWaits.await();  //因为是向队尾添加元素，所以用tailWaits
    38.             }
    39.             //当队列不满时，可以进行添加
    40.             array[tail] = e;
    41.             //先判断tail是否越界
    42.             if(++tail == array.length){
    43.                 tail = 0;
    44.             }
    45.             //让size+1
    46.             size++;
    47.             //这时要通知poll，因为放入了一个元素，所以队列肯定不为空，通知poll线程可以取元素了
    48.             headWaits.signal(); //唤醒poll线程
    49.         }finally {
    50.             //释放锁
    51.             reentrantLock.unlock();
    52.         }
    53.     }
    54.  
    55.     /**
    56.      * 向队尾添加元素，加入超时时间，如果队满，并且过了超时时间，返回false
    57.      * @param e 要添加的元素
    58.      * @param timeout 超时时间
    59.      * @return 是否添加成功
    60.      * @throws InterruptedException
    61.      */
    62.     @Override
    63.     public boolean offer(E e, long timeout) throws InterruptedException {
    64.         //先加锁
    65.         reentrantLock.lock();
    66.  
    67.         try{
    68.             //将传入的时间转换为纳秒
    69.             long t = TimeUnit.MILLISECONDS.toNanos(timeout);
    70.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    71.             while (isFull()){
    72.                 /**
    73.                  *  如果是满的，就让当前线程阻塞
    74.                  *  我们用加入时间的方法来阻塞
    75.                  *  注意这里我们每次阻塞完唤醒后，就更新等待时间
    76.                  *  如果要等待的时间是5s,那么如果等待了1s发现队列有空，那么唤醒之后，
    77.                  *  如果是虚假唤醒，就要再次等待，那么下次等待时间就是4s
    78.                  *  如果等待时间 t<=0了，说明等待超时，直接返回false，不要在等了
    79.                  **/
    80.                 if(t <= 0){
    81.                     return false;
    82.                 }
    83.                 t = tailWaits.awaitNanos(t);
    84.             }
    85.             //当队列不满时，可以进行添加
    86.             array[tail] = e;
    87.             //先判断tail是否越界
    88.             if(++tail == array.length){
    89.                 tail = 0;
    90.             }
    91.             //让size+1
    92.             size++;
    93.             //这时要通知poll，因为放入了一个元素，所以队列肯定不为空，通知poll线程可以取元素了
    94.             headWaits.signal(); //唤醒poll线程
    95.             /**
    96.              * 到这里就说明添加成功，返回true
    97.              */
    98.             return true;
    99.         }finally {
    100.             //释放锁
    101.             reentrantLock.unlock();
    102.         }
    103.     }
    104.  
    105.     /**
    106.      * 移除队头元素，如果队列为空，则阻塞当前线程
    107.      * @return 队头元素
    108.      * @throws InterruptedException
    109.      */
    110.     @Override
    111.     public E poll() throws InterruptedException {
    112.         //先加锁
    113.         reentrantLock.lock();
    114.  
    115.         try {
    116.           //先循环判断队列是否为空
    117.             while (isEmpty()){
    118.                 //如果队列为空，让当前线程阻塞
    119.                 headWaits.await();
    120.             }
    121.             //如果队列不为空，可以取了
    122.             E e = array[head];
    123.             array[head] = null; //help GC
    124.             //判断head是否越界
    125.             if(++head == array.length){
    126.                 head = 0;
    127.             }
    128.             //让size-1
    129.             size--;
    130.             //这时队列因为取出了一个元素，所以肯定不为满，通知offer线程可以添加元素了
    131.             tailWaits.signal(); //唤醒offer线程
    132.             return e;
    133.         }finally {
    134.             //释放锁
    135.             reentrantLock.unlock();
    136.         }
    137.     }
    138.  
    139.     /**
    140.      * 移除队头元素，加入超时时间，如果队空，并且超过等待时间，返回null
    141.      * @param timeout 超时时间
    142.      * @return 队头元素
    143.      * @throws InterruptedException
    144.      */
    145.     @Override
    146.     public E poll(long timeout) throws InterruptedException {
    147.         //先加锁
    148.         reentrantLock.lock();
    149.  
    150.         try {
    151.             long t = TimeUnit.MILLISECONDS.toNanos(timeout);
    152.             //先循环判断队列是否为空
    153.             while (isEmpty()){
    154.                 //如果队列为空，让当前线程阻塞
    155.                 if(t <= 0){
    156.                     return null;
    157.                 }
    158.                  t = headWaits.awaitNanos(t);
    159.             }
    160.             //如果队列不为空，可以取了
    161.             E e = array[head];
    162.             array[head] = null; //help GC
    163.             //判断head是否越界
    164.             if(++head == array.length){
    165.                 head = 0;
    166.             }
    167.             //让size-1
    168.             size--;
    169.             //这时队列因为取出了一个元素，所以肯定不为满，通知offer线程可以添加元素了
    170.             tailWaits.signal(); //唤醒offer线程
    171.             return e;
    172.         }finally {
    173.             //释放锁
    174.             reentrantLock.unlock();
    175.         }
    176.     }
    177.  
    178.     @Override
    179.     public boolean isFull() {
    180.         return size == array.length;
    181.     }
    182.  
    183.     @Override
    184.     public boolean isEmpty() {
    185.         return size == 0;
    186.     }
    187.  
    188.  
    189.     @Override
    190.     public String toString() {
    191.         return Arrays.toString(array);
    192.     }
    193. }

    3.解释说明

    注释中也都写了，我在这里再次解释一下:

    (1).offer添加元素

      当我们执行offer方法向队列中添加元素时:

     1. 我们需要先加锁，

     2. 我们判断队列是否为满，

     3.  如果满了，我们需要让当前线程阻塞 ,

      注意:这里使用的是while循环，为什么不用if呢?我们想，如果用if，那么它只会判断一次，如果当某个时刻，队列从满变为不满，这时我们阻塞的offer线程被唤醒，将要去添加元素，但就在此时，另一个offer1线程可能在offer线程添加之前抢先往队列中添加了元素，那么offer线程再去添加就会报错，也就是虚假唤醒（spurious wakeups），使用while循环判断队列是否为满，可以在阻塞线程被唤醒后重新判断队列是否满足条件。如果队列仍然满，线程会继续被阻塞，直到队列有空闲位置。这样可以预防虚假唤醒的问题，确保线程只有在满足条件的情况下才会执行添加元素的操作。

     4.当while条件不成立时，也就是队列有空为，并且此时没有其他线程来争抢，那么就可以往队列中添加元素了 ， queue[tail] = e;

    5.判断++tail是否达到了数组末尾位置，如果到了，那么重新调整为0，相当于一个圆圈

    6.让size++

    7.当offer线程向队列中添加元素后，此时队列肯定不为空，我们应该向poll线程发出信号，可以唤醒，相当于操作系统中的信号量机制

    （2）poll取出元素

        1.先加锁

        2.判断队列是否为空，同理使用while循环判断

        3.当队列不为空时，取出队头元素

        4.判断++head是否到数组末尾位置，如果到了，重新置为0

        5.让size--;

        6.当poll线程从队列中取出元素后，队列肯定不为满，我们应该向offer线程发出信号，唤醒offer线程

    4.timeout超时时间

     我们可以为offer和poll设置超时时间,当超过了等待时间，将直接返回，不在执行

     看一下offer方法设置timeout

        /**
                     *  如果是满的，就让当前线程阻塞
                     *  我们用加入时间的方法来阻塞
                     *  注意这里我们每次阻塞完唤醒后，就更新等待时间
                     *  如果要等待的时间是5s,那么如果等待了1s发现队列有空，那么唤醒之后，
                     *  如果是虚假唤醒，就要再次等待，那么下次等待时间就是4s
                     *  如果等待时间 t<=0了，说明等待超时，直接返回false，不要在等了
                     **/

    poll方法同理... 

    5.测试

     下面让我们来测试一下代码

      

    1. public class TestBlockQueue {
    2.  
    3.  
    4.     public static void main(String[] args) throws InterruptedException {
    5.  
    6.  
    7.         ArrayBlockQueue queue = new ArrayBlockQueue<>(4);
    8.  
    9.         queue.offer("task1");
    10.         queue.offer("task2");
    11.         queue.offer("task3");
    12.         queue.offer("task4");
    13.  
    14.         System.out.println(queue);
    15.  
    16.         new Thread(() -> {
    17.             System.out.println(Thread.currentThread().getName() + "开始添加元素...");
    18.             try {
    19.                 boolean flag = queue.offer("task5", 4000);
    20.                 if(flag){
    21.                     System.out.println(Thread.currentThread().getName() + "添加元素成功....");
    22.                 }else {
    23.                     System.out.println(Thread.currentThread().getName()+"添加元素超超时失败....");
    24.                 }
    25.  
    26.                 System.out.println(queue);
    27.             } catch (InterruptedException e) {
    28.                 throw new RuntimeException(e);
    29.             }
    30.  
    31.         }, "t1").start();
    32.  
    33.  
    34.  
    35.     }
    36. }

     我们给队列初始大小设为4，然后向队列中添加4个任务，此时队列为满，然后我们开启一个t1线程向队列中添加元素，设置超时时间为4s,判断是否能添加成功

    运行:

    1. [task1, task2, task3, task4]
    2. t1开始添加元素...
    3. t1添加元素超超时失败....
    4. [task1, task2, task3, task4]
    5.  
    6. 进程已结束,退出代码0

    可以看到，过了4s，添加失败，因为我们并没有取出任何元素，所以offer线程一直阻塞直到超时失败！

    下面，我们让主线程先休眠2s,然后取出一个元素，再次观察t1是否能添加成功：

    1.         //让主线程休眠2s，然后poll
    2.         Thread.sleep(2000);
    3.         queue.poll();

    在上面的代码中添加以上代码，然后运行：

    1. [task1, task2, task3, task4]
    2. t1开始添加元素...
    3. t1添加元素成功....
    4. [task5, task2, task3, task4]
    5.  
    6. 进程已结束,退出代码0

    这次可以看到，task5被成功的添加到了队列中，因为我们在超时时间之前取出了队列的一个元素，队列有了空位，task5就可以添加到队列中了。

    五.基于数组实现双锁的阻塞队列

    1.问题

     上面我们是用一把锁来给offer线程和poll线程加锁，他们两个操作用的同一把锁，这样其实并不好，效率比较低，而且添加和取出操作应该是两个互不影响的操作，是互相解耦的，所以我们应该使用双锁来给他们分别加锁和释放锁

    2.关于size共享变量

      当我们换成双锁后，需要思考一个问题，这个头指针head和尾指针tail，head是poll线程用来取出队列头元素的，tail是offer线程用来向队尾添加元素使用的，所以说head和tail这两个变量是互不影响的，它们分别在各自的线程里使用，但是对于size，在offer线程中，最后要让size++;在poll线程中，最后要让size--,这就是共享的变量了，可能会出现线程安全问题，如果两个线程不是顺序执行的，而是交错执行，就会是size的值发生混乱,所以我们要对size作约束

     在java中,我们可以实现原子类来对变量进行线程安全保护，对于int类型的我们使用AtomicInteger，

    加一可以使用getAndIncreament()方法，减一可以使用getAndDecreament()方法

    1.     //offer锁
    2.     private final ReentrantLock headLock = new ReentrantLock();
    3.     //poll锁
    4.     private final ReentrantLock tailLock = new ReentrantLock();
    5.     //条件对象
    6.     private final Condition headWaits = headLock.newCondition();
    7.     private final Condition tailWaits = tailLock.newCondition();

    3.死锁问题

      如果我们添加了双锁，那么我们需要设置各自的条件去阻塞和唤醒线程，先看一下offer线程,

    1.     @Override
    2.     public void offer(E e) throws InterruptedException {
    3.  
    4.        
    5.  
    6.         //先加锁
    7.         tailLock.lock();
    8.         try{
    9.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    10.             while (isFull()){
    11.                 //如果是满的，就让当前线程阻塞
    12.                 tailWaits.await();  //因为是向队尾添加元素，所以用tailWaits
    13.             }
    14.             //当队列不满时，可以进行添加
    15.             array[tail] = e;
    16.             //先判断tail是否越界
    17.             if(++tail == array.length){
    18.                 tail = 0;
    19.             }
    20.             //让size+1
    21.             size.getAndIncrement();
    22.  
    23.              headLock.lock();
    24.             try {
    25.               
    26.                 headWaits.signal(); //poll1
    27.             }finally {
    28.                 headLock.unlock();
    29.             }
    30.             
    31.         }finally {
    32.             //释放锁
    33.             tailLock.unlock();
    34.         }
    35.  
    36.        
    37.         }
    38.  
    39.     }

    注意看，这里我们用headWaits唤醒线程时，它是在tailLock释放锁之前，也就是一个嵌套结构，这样就会导致死锁的发生,

     

    如果tailLock先加锁了，然后headLock也去加锁，之后在offer线程中的headLock想去加锁就加不上了，同理poll线程中的tailLock想加锁也加不上去，他们两个线程互相僵持，陷入了死锁状态，为了防止死锁发生，我们只要把嵌套结构改为平级结构就可以了，这样就能保证一定是释放完锁之后再去加锁，一定可以加锁成功！

     

    1.    @Override
    2.     public void offer(E e) throws InterruptedException {
    3.  
    4.        
    5.  
    6.         //先加锁
    7.         tailLock.lock();
    8.         try{
    9.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    10.             while (isFull()){
    11.                 //如果是满的，就让当前线程阻塞
    12.                 tailWaits.await();  //因为是向队尾添加元素，所以用tailWaits
    13.             }
    14.             //当队列不满时，可以进行添加
    15.             array[tail] = e;
    16.             //先判断tail是否越界
    17.             if(++tail == array.length){
    18.                 tail = 0;
    19.             }
    20.             //让size+1
    21.             size.getAndIncrement();
    22.  
    23.           
    24.         }finally {
    25.             //释放锁
    26.             tailLock.unlock();
    27.         }
    28.  
    29.  
    30.        /**
    31.          * 平级可以防止死锁
    32.          * 唤醒poll线程,但是因为是各自不同的锁，所以需要在他们各自的锁内唤醒
    33.          */  
    34.            headLock.lock();
    35.             try {
    36.               
    37.                 headWaits.signal(); //poll1
    38.             }finally {
    39.                 headLock.unlock();
    40.             }
    41.             
    42.  
    43.     }

    这样的平级结构就可以防止死锁发生了... 

     4.级联唤醒

      以上的唤醒逻辑，是每次都要进行一次唤醒，这样其实效率还不是最好的，那么我们可不可以减少唤醒的次数来提高效率呢？我们可以通过级联唤醒来实现

    （1）offer中只唤醒一次，其他交给poll线程唤醒

      先看一下offer线程,在最后要随机唤醒一个poll线程,我们让它只唤醒一个线程，剩下的线程交给poll线程自己去唤醒，比如有poll1,poll2,poll3，poll4四个线程，那么我们的offer线程只唤醒poll1线程，然后让poll1去唤醒poll2,poll2去唤醒poll3,以此类推...,这样就可以让offer只执行一次唤醒，提高了效率

     

    1.   @Override
    2.     public void offer(E e) throws InterruptedException {
    3.  
    4.         //记录一下每次size加一之前的值
    5.         int c;
    6.  
    7.         //先加锁
    8.         tailLock.lock();
    9.         try{
    10.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    11.             while (isFull()){
    12.                 //如果是满的，就让当前线程阻塞
    13.                 tailWaits.await();  //因为是向队尾添加元素，所以用tailWaits
    14.             }
    15.             //当队列不满时，可以进行添加
    16.             array[tail] = e;
    17.             //先判断tail是否越界
    18.             if(++tail == array.length){
    19.                 tail = 0;
    20.             }
    21.             //让size+1
    22.             c = size.getAndIncrement();
    23.             //如果c+1>array.length,说明队列还是有空位置，就自己唤醒后面的线程
    24.             if( c+1 > array.length){
    25.                 tailWaits.signal();
    26.             }
    27.         }finally {
    28.             //释放锁
    29.             tailLock.unlock();
    30.         }
    31.  
    32.         /**
    33.          * 平级可以防止死锁
    34.          * 唤醒poll线程,但是因为是各自不同的锁，所以需要在他们各自的锁内唤醒
    35.          */
    36.         if( c == 0){   //offer_0,offer_1,offer_2, c=0说明是第一个offer_0,队列时空的，准备添加第一个
    37.             headLock.lock();
    38.             try {
    39.                 /**
    40.                  * 这里我们只唤醒第一个poll1线程，其他的
    41.                  * 交给poll线程自己唤醒
    42.                  * 如果c为0,说明队列为空，准备要添加第一个元素，就只让offer_0来唤醒
    43.                  */
    44.                 headWaits.signal(); //poll1
    45.             }finally {
    46.                 headLock.unlock();
    47.             }
    48.         }
    49.  
    50.     }

       我们来看改进后的代码，先设置一个c变量来记录每次size改变前的值,在唤醒时，先判断c是否等于0,这里等于0说明队列从空的状态开始，去添加第一个元素，那么也就是第一个offer1线程，让它去唤醒poll线程，其他的poll线程交给poll线程自己去唤醒，再看一下poll的代码:

    1.  @Override
    2.     public E poll() throws InterruptedException {
    3.         E e;
    4.         int c;
    5.         //先加锁
    6.         headLock.lock();
    7.         try {
    8.             //先循环判断队列是否为空
    9.             while (isEmpty()){
    10.                 //如果队列为空，让当前线程阻塞
    11.                 headWaits.await();
    12.             }
    13.             //如果队列不为空，可以取了
    14.              e = array[head];
    15.             array[head] = null; //help GC
    16.             //判断head是否越界
    17.             if(++head == array.length){
    18.                 head = 0;
    19.             }
    20.             //让size-1
    21.             c = size.getAndDecrement();
    22.             //在这里让poll1唤醒poll2,poll2接着唤醒poll3...
    23.             if( c > 1 ){
    24.                 //c>1说明队列中还有不止一个元素，可以继续唤醒其他poll线程来去元素
    25.                 headWaits.signal();
    26.             }
    27.  
    28.         }finally {
    29.             //释放锁
    30.             headLock.unlock();
    31.         }
    32.  
    33.         /**
    34.          * 平级，防止死锁
    35.          * 唤醒offer线程,需要加锁
    36.          *  当 c == array.length时，说明队列从满变为不满，这时才去
    37.          *  给tailLock加锁
    38.          */
    39.         if( c == array.length ){
    40.             tailLock.lock();
    41.             try {
    42.                 tailWaits.signal();
    43.             }finally {
    44.                 tailLock.unlock();
    45.             }
    46.         }
    47.  
    48.         return e;
    49.     }

    在poll代码中,只需要判断c是否大于1，如果c>1,说明队列中还有元素，可以继续唤醒，那么就让poll1去唤醒poll2,poll2去唤醒poll3.....

    (2)poll中只唤醒一次，其他交给offer线程唤醒

    再来看poll中，当c==array.length时，说明这时是队列从满变为不满，只有这时才去唤醒，其他情况，比如队列时不满的，也不去唤醒，

    然后在offer中,当c+1

    5.完整代码

    1. /**
    2.  * 基于数组的双锁实现阻塞队列
    3.  */
    4. public class ArrayDLBlockQueue implements BlockingQueue {
    5.  
    6.  
    7.  
    8.     private final E[] array;
    9.     private int head;
    10.     private int tail;
    11.     //在双锁条件下，size是共享的变量，需要保证原子性
    12.     private AtomicInteger size = new AtomicInteger();
    13.  
    14.     @SuppressWarnings("all")
    15.     public ArrayDLBlockQueue(int capacity){
    16.         array = (E[]) new Object[capacity];
    17.     }
    18.  
    19.     //offer锁
    20.     private final ReentrantLock headLock = new ReentrantLock();
    21.     //poll锁
    22.     private final ReentrantLock tailLock = new ReentrantLock();
    23.     //条件对象
    24.     private final Condition headWaits = headLock.newCondition();
    25.     private final Condition tailWaits = tailLock.newCondition();
    26.  
    27.     /**
    28.      * 向队尾添加元素，如果队列为满，则阻塞当前线程
    29.      * @param e 要添加元素
    30.      * @throws InterruptedException
    31.      */
    32.     @Override
    33.     public void offer(E e) throws InterruptedException {
    34.  
    35.         //记录一下每次size加一之前的值
    36.         int c;
    37.  
    38.         //先加锁
    39.         tailLock.lock();
    40.         try{
    41.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    42.             while (isFull()){
    43.                 //如果是满的，就让当前线程阻塞
    44.                 tailWaits.await();  //因为是向队尾添加元素，所以用tailWaits
    45.             }
    46.             //当队列不满时，可以进行添加
    47.             array[tail] = e;
    48.             //先判断tail是否越界
    49.             if(++tail == array.length){
    50.                 tail = 0;
    51.             }
    52.             //让size+1
    53.             c = size.getAndIncrement();
    54.             //如果c+1>array.length,说明队列还是有空位置，就自己唤醒后面的线程
    55.             if( c+1 > array.length){
    56.                 tailWaits.signal();
    57.             }
    58.         }finally {
    59.             //释放锁
    60.             tailLock.unlock();
    61.         }
    62.  
    63.         /**
    64.          * 平级可以防止死锁
    65.          * 唤醒poll线程,但是因为是各自不同的锁，所以需要在他们各自的锁内唤醒
    66.          */
    67.         if( c == 0){   //offer_0,offer_1,offer_2, c=0说明是第一个offer_0,队列时空的，准备添加第一个
    68.             headLock.lock();
    69.             try {
    70.                 /**
    71.                  * 这里我们只唤醒第一个poll1线程，其他的
    72.                  * 交给poll线程自己唤醒
    73.                  * 如果c为0,说明队列为空，准备要添加第一个元素，就只让offer_0来唤醒
    74.                  */
    75.                 headWaits.signal(); //poll1
    76.             }finally {
    77.                 headLock.unlock();
    78.             }
    79.         }
    80.  
    81.     }
    82.  
    83.     /**
    84.      * 向队尾添加元素，加入超时时间，如果队满，并且过了超时时间，返回false
    85.      * @param e 要添加的元素
    86.      * @param timeout 超时时间
    87.      * @return 是否添加成功
    88.      * @throws InterruptedException
    89.      */
    90.     @Override
    91.     public boolean offer(E e, long timeout) throws InterruptedException {
    92.         int c;
    93.         //先加锁
    94.         tailLock.lock();
    95.  
    96.         try{
    97.             //将传入的时间转换为纳秒
    98.             long t = TimeUnit.MILLISECONDS.toNanos(timeout);
    99.             //判断队列是否为满，这里要用while循环来判断，防止虚假唤醒
    100.             while (isFull()){
    101.                 /**
    102.                  *  如果是满的，就让当前线程阻塞
    103.                  *  我们用加入时间的方法来阻塞
    104.                  *  注意这里我们每次阻塞完唤醒后，就更新等待时间
    105.                  *  如果要等待的时间是5s,那么如果等待了1s发现队列有空，那么唤醒之后，
    106.                  *  如果是虚假唤醒，就要再次等待，那么下次等待时间就是4s
    107.                  *  如果等待时间 t<=0了，说明等待超时，直接返回false，不要在等了
    108.                  **/
    109.                 if(t <= 0){
    110.                     return false;
    111.                 }
    112.                 t = tailWaits.awaitNanos(t);
    113.             }
    114.             //当队列不满时，可以进行添加
    115.             array[tail] = e;
    116.             //先判断tail是否越界
    117.             if(++tail == array.length){
    118.                 tail = 0;
    119.             }
    120.             //让size+1
    121.             c = size.getAndIncrement();
    122.             if(c+1 < array.length){
    123.                 tailWaits.signal();
    124.             }
    125.         }finally {
    126.             //释放锁
    127.             tailLock.unlock();
    128.         }
    129.  
    130.         /**
    131.          * 平级防止死锁
    132.          */
    133.         if( c==0){
    134.             headLock.lock();
    135.             try {
    136.                 headWaits.signal();
    137.             }finally {
    138.                 headLock.unlock();
    139.             }
    140.         }
    141.  
    142.  
    143.         return true;
    144.     }
    145.  
    146.     /**
    147.      * 移除队头元素，如果队列为空，则阻塞当前线程
    148.      * @return 队头元素
    149.      * @throws InterruptedException
    150.      */
    151.     @Override
    152.     public E poll() throws InterruptedException {
    153.         E e;
    154.         int c;
    155.         //先加锁
    156.         headLock.lock();
    157.         try {
    158.             //先循环判断队列是否为空
    159.             while (isEmpty()){
    160.                 //如果队列为空，让当前线程阻塞
    161.                 headWaits.await();
    162.             }
    163.             //如果队列不为空，可以取了
    164.              e = array[head];
    165.             array[head] = null; //help GC
    166.             //判断head是否越界
    167.             if(++head == array.length){
    168.                 head = 0;
    169.             }
    170.             //让size-1
    171.             c = size.getAndDecrement();
    172.             //在这里让poll1唤醒poll2,poll2接着唤醒poll3...
    173.             if( c > 1 ){
    174.                 //c>1说明队列中还有不止一个元素，可以继续唤醒其他poll线程来去元素
    175.                 headWaits.signal();
    176.             }
    177.  
    178.         }finally {
    179.             //释放锁
    180.             headLock.unlock();
    181.         }
    182.  
    183.         /**
    184.          * 平级，防止死锁
    185.          * 唤醒offer线程,需要加锁
    186.          *  当 c == array.length时，说明队列从满变为不满，这时才去
    187.          *  给tailLock加锁
    188.          */
    189.         if( c == array.length ){
    190.             tailLock.lock();
    191.             try {
    192.                 tailWaits.signal();
    193.             }finally {
    194.                 tailLock.unlock();
    195.             }
    196.         }
    197.  
    198.         return e;
    199.     }
    200.  
    201.     /**
    202.      * 移除队头元素，加入超时时间，如果队空，并且超过等待时间，返回null
    203.      * @param timeout 超时时间
    204.      * @return 队头元素
    205.      * @throws InterruptedException
    206.      */
    207.     @Override
    208.     public E poll(long timeout) throws InterruptedException {
    209.         E e;
    210.         int c;
    211.         //先加锁
    212.         headLock.lock();
    213.  
    214.         try {
    215.             long t = TimeUnit.MILLISECONDS.toNanos(timeout);
    216.             //先循环判断队列是否为空
    217.             while (isEmpty()){
    218.                 //如果队列为空，让当前线程阻塞
    219.                 if(t <= 0){
    220.                     return null;
    221.                 }
    222.                 t = headWaits.awaitNanos(t);
    223.             }
    224.             //如果队列不为空，可以取了
    225.              e = array[head];
    226.             array[head] = null; //help GC
    227.             //判断head是否越界
    228.             if(++head == array.length){
    229.                 head = 0;
    230.             }
    231.             //让size-1
    232.             c = size.getAndDecrement();
    233.             if( c > 1){
    234.                 headWaits.signal();
    235.             }
    236.         }finally {
    237.             //释放锁
    238.             headLock.unlock();
    239.         }
    240.  
    241.         /**
    242.          * 平级，防止死锁
    243.          * 唤醒offer线程,需要加锁
    244.          */
    245.         if( c == array.length ){
    246.             tailLock.lock();
    247.             try {
    248.                 tailWaits.signal();
    249.             }finally {
    250.                 tailLock.unlock();
    251.             }
    252.         }
    253.        
    254.         return e;
    255.     }
    256.  
    257.     @Override
    258.     public boolean isFull() {
    259.         return size.get() == array.length;
    260.     }
    261.  
    262.     @Override
    263.     public boolean isEmpty() {
    264.         return size.get() == 0;
    265.     }
    266.  
    267.  
    268.     @Override
    269.     public String toString() {
    270.         return Arrays.toString(array);
    271.     }
    272.  
    273.

    以上就是对阻塞队列的分析了，读者可以在此基础上实现链表的实现等,我们下期再见!

  • 相关阅读:
    LQ0112 立方和【进制】
    Redis分布式锁
    由ASP.NET Core根据路径下载文件异常引发的探究
    SpringBoot读取配置的方式
    Python爬虫入门
    Jenkis 配置钉钉通知
    C语言对单链表所有操作与一些相关面试题
    机器人课程教师面对的困境有哪些
    接口面试题
    自定义一个中间件（功能：解析POST提交到服务器的表单数据）
  • 原文地址：https://blog.csdn.net/jjhnb123/article/details/134340612