JUC并发编程学习笔记（九）阻塞队列

2023-11-04 18:53 由高同学，你好发表于 #后端开发

阻塞队列

阻塞

队列

队列的特性：FIFO（fist inpupt fist output）先进先出

不得不阻塞的情况

什么情况下会使用阻塞队列：多线程并发处理、线程池

学会使用队列

添加、移除

四组API

方式	抛出异常	不抛出异常，有返回值	阻塞等待	超时等待
添加	add	offer	put	offer(E e, long timeout, TimeUnit unit)
移除	remove	poll	take	poll(long timeout, TimeUnit unit)
检测队首元素	element	peek

1、抛出异常

public static void test01(){
    //队列是有大小的，创建时要明确该队列最大能有几个元素
    BlockingQueue queue = new ArrayBlockingQueue<>(3);
    //当你添加的元素超出队列规定的最大元素量时，抛出异常java.lang.IllegalStateException: Queue full
    System.out.println(queue.add("A"));
    System.out.println(queue.add("B"));
    System.out.println(queue.add("C"));
    //当你移除的元素超出队列中含有的元素量时，抛出异常java.util.NoSuchElementException
    System.out.println(queue.element());
    System.out.println(queue.remove());
    System.out.println(queue.element());
    System.out.println(queue.remove());
    System.out.println(queue.element());
    System.out.println(queue.remove());


    System.out.println(queue.add("D"));
}

2、不抛出异常

public static void test02(){
    //队列是有大小的，创建时要明确该队列最大能有几个元素
    BlockingQueue queue = new ArrayBlockingQueue<>(3);
    //当你添加的元素超出队列规定的最大元素量时，不抛出异常，通过返回的boolean值为false来表明
    System.out.println(queue.offer("A"));
    System.out.println(queue.offer("B"));
    System.out.println(queue.offer("C"));
    //当你移除的元素超出队列中含有的元素量时，不抛出异常，返回null来提示队列已经没有元素可以弹出了
    System.out.println(queue.peek());
    System.out.println(queue.poll());
    System.out.println(queue.peek());
    System.out.println(queue.poll());
    System.out.println(queue.peek());
    System.out.println(queue.poll());


    System.out.println(queue.offer("D"));
}

3、阻塞等待

    public static void test03() throws InterruptedException {
        //队列是有大小的，创建时要明确该队列最大能有几个元素
        BlockingQueue queue = new ArrayBlockingQueue<>(3);

        queue.put("a");
        queue.put("b");
        queue.put("c");
        //当队列被元素占满后再进行添加，会一直阻塞等待，知道该元素加入队列
//        queue.put("c");
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
        //当队列中没有元素时还去取，也会一直阻塞等待
//        System.out.println(queue.take());

    }

4、超时等待

public static void test04() throws Exception{
    //队列是有大小的，创建时要明确该队列最大能有几个元素
    BlockingQueue queue = new ArrayBlockingQueue<>(3);
    System.out.println(queue.offer("a"));
    System.out.println(queue.offer("b"));
    System.out.println(queue.offer("c"));
    //当队列被占满时，设置超时等待，超过两秒就不等了,进不去就不进了
    System.out.println(queue.offer("d", 2, TimeUnit.SECONDS));
    System.out.println(queue.poll());
    System.out.println(queue.poll());
    System.out.println(queue.poll());
    //当队列元素为空时，设置超时等待，超过两秒就不等了,拿不到就不拿了
    System.out.println(queue.poll(2, TimeUnit.SECONDS));
}

SynchronousQueue

同步队列：同步队列每次插入操作都必须等待另一个线程执行相应的删除操作，反之亦然。同步队列没有容量，也就意味着只有当一个元素弹出后才能再进入一个元素，只能同时有一个元素。

存取：put、take

package org.example.qu;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

public class SynchronousQueueDemo {
    public static void main(String[] args) {
        SynchronousQueue<String> synchroQueue = new SynchronousQueue<String>();

        new Thread(() -> {
            try {
                //直接在队列中存入三个元素
                System.out.println(Thread.currentThread().getName() + ": put 1");
                synchroQueue.put("1");
                System.out.println(Thread.currentThread().getName() + ": put 2");
                synchroQueue.put("2");
                System.out.println(Thread.currentThread().getName() + ": put 3");
                synchroQueue.put("3");


            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "T1").start();
        new Thread(() -> {
            try {
                //每次取之间延迟三秒
                TimeUnit.SECONDS.sleep(3);
                System.out.println(synchroQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(synchroQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(synchroQueue.take());

            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "T2").start();

    }
}

我们通过结果发现，将要存入的元素会等待已经被存入的元素取出后才会存入，延迟三秒取出时，将要存入的元素也在等待三秒取出后再存入，这就是所谓的同步队列每次插入操作都必须等待另一个线程执行相应的删除操作，反之亦然。

同步队列和其它BlockingQueue不一样，他不存储元素，你put了一个值必须take取出，否则不能再put进去元素。