JUC-08-队列

• 队列：先入先出的数据结构
• 阻塞：写入阻塞和读取阻塞

1. 阻塞队列

• jdk文档中的介绍

• BlockingQueue(一般用于线程池)

• Queue的家族结构如图所示

2. 阻塞队列 四组API

• 抛出异常
• 不会抛出异常
• 阻塞等待
• 超时等待

方式	抛出异常	有返回值,不抛出异常	阻塞等待	超时等待
添加	add()	offer()	put()	offer(“加入的内容”,seconds,TimeUnit.Seconds)
移除	remove()	poll()	take()	poll(seconds,TimeUnit.Seconds)
判断队列首	element()	peek()	-	-

1. 抛出异常

public class TestBq {
    public static void main(String[] args) {
        /*
        抛出异常
         */
        test1();

    }


    public static void test1(){
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
        System.out.println(blockingQueue.add("a"));
        System.out.println(blockingQueue.add("b"));
        System.out.println(blockingQueue.add("c"));


//        // 抛出异常java.lang.IllegalStateException: Queue full
//        System.out.println(blockingQueue.add("d"));

        System.out.println("===============");

        // 移除操作
        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());

        // 队列为空之后 报错 java.util.NoSuchElementException
        System.out.println(blockingQueue.remove());
    }
}

2. 不抛出异常,有返回值

public static void test2(){
    ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);

    System.out.println(blockingQueue.offer("a"));
    System.out.println(blockingQueue.offer("b"));
    System.out.println(blockingQueue.offer("c"));
    System.out.println(blockingQueue.offer("d")); // 不抛出异常

    System.out.println("===============");
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());  // null
}


public static void main(String[] args) {
    // 不抛出异常
    test2();
}

3. 一直阻塞

public static void test3() throws InterruptedException {
    // 等待阻塞
    ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
    // 一直阻塞
    blockingQueue.put("a");
    blockingQueue.put("b");
    blockingQueue.put("c");

    // 队列没有位置，一直等待
    blockingQueue.put("d");


    System.out.println("===============");
    System.out.println(blockingQueue.take());
    System.out.println(blockingQueue.take());
    System.out.println(blockingQueue.take());
}

public static void main(String[] args) throws InterruptedException {

    // 一直阻塞
    test3();
}

4. 超时退出

public static void test4() throws InterruptedException {
    //超时等待
    ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);

    blockingQueue.offer("a");
    blockingQueue.offer("b");
    blockingQueue.offer("c");
    blockingQueue.offer("d", 2,TimeUnit.SECONDS); // 如果加d，只能2秒，超时就退出

    System.out.println("===============");
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS));
}


public static void main(String[] args) throws InterruptedException {
    //超时等待
    test4();
}

3. 同步队列

• 阻塞队列：进去一个元素，必须等待取出来之后

• 同步队列：拿一个取一个。

  同步队列:和其他的阻塞队列是不一样的，
  SynchronousQueue是不存储元素的，
  只要put了一个元素，必须先take出来，否则不能继续put值

举例描述：

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

public class TestSynchronizedQueue {
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>(); // 同步队列

        // 一个线程在put
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+"put 1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+"put 2");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+"put 3");
                blockingQueue.put("1");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t1").start();


        // 一个线程在take
        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+"get "+blockingQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+"get "+blockingQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+"get "+blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t2").start();
    }
}

打赏