JUC基础之阻塞队列四组API,SynchronousQueue同步队列

队列与阻塞

阻塞队列：

BlokingQueue

API

主要的Queue的实现类：

AbstractQueue：非阻塞队列
BlockingQueue：阻塞队列
Deque：双端队列

看一下智慧树

智慧树

什么情况下我们要使用阻塞队列？

多线程并发，线程池

BlockingQueue 四组API

操作	抛出异常	不抛出异常，有返回值	阻塞等待	超时等待
添加	add()	offer()	put()	offer(e,timeout,TimeUnit)
删除	remove()	pull()	take()	pull(e,timeout,TimeUnit)
检查队列首位元素	element(),peek()	-	-	-

抛出异常：

/*
抛出异常
 */
 public static  void testThrow()
 {
     //   初始化必须 设置队列的大小
     ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(3);


     System.out.println(arrayBlockingQueue.add("Y1"));
     System.out.println(arrayBlockingQueue.add("Y2"));
     System.out.println(arrayBlockingQueue.add("Y3"));


     //抛出异常 Exception in thread "main" java.lang.IllegalStateException: Queue full
     //System.out.println(arrayBlockingQueue.add("Y4"));


     System.out.println(arrayBlockingQueue.remove());
     System.out.println(arrayBlockingQueue.remove());
     System.out.println(arrayBlockingQueue.remove());


     //抛出异常 Exception in thread "main" java.util.NoSuchElementException
     //System.out.println(arrayBlockingQueue.remove());
 }

不抛异常，有返回值:

/*
不抛出异常，有返回值
*/
public static void testNoThrow()
{
    ArrayBlockingQueue<Object> arrayBlockingQueue = new ArrayBlockingQueue<>(3);


    System.out.println(arrayBlockingQueue.offer("Y1"));
    System.out.println(arrayBlockingQueue.offer("Y2"));
    System.out.println(arrayBlockingQueue.offer("Y3"));
    //不抛出异常 返回false
    //System.out.println(arrayBlockingQueue.offer("Y4"));


    System.out.println(arrayBlockingQueue.poll());
    System.out.println(arrayBlockingQueue.poll());
    System.out.println(arrayBlockingQueue.poll());
    //不抛出异常 返回null
    System.out.println(arrayBlockingQueue.poll());
}

阻塞等待:

/*
阻塞，(一直阻塞)
 */
public static void block() throws InterruptedException
{
    ArrayBlockingQueue<Object> arrayBlockingQueue = new ArrayBlockingQueue<>(3);


    arrayBlockingQueue.put("Y1");
    arrayBlockingQueue.put("Y1");
    arrayBlockingQueue.put("Y1");
    //一直阻塞,一直等待进入
    arrayBlockingQueue.put("Y1");


    System.out.println(arrayBlockingQueue.take());
    System.out.println(arrayBlockingQueue.take());
    System.out.println(arrayBlockingQueue.take());
    //一直等待取出
    System.out.println(arrayBlockingQueue.take());
    
}

超时等待:

/*
阻塞（等待超时）
 */
public static void blockTimeOut() throws InterruptedException
{
    ArrayBlockingQueue<Object> arrayBlockingQueue = new ArrayBlockingQueue<>(3);


    arrayBlockingQueue.offer("Y1");
    arrayBlockingQueue.offer("Y1");
    arrayBlockingQueue.offer("Y1");
    //存入操作  等待超过2秒就退出
    arrayBlockingQueue.offer("Y1",2, TimeUnit.SECONDS);
    
    arrayBlockingQueue.poll();
    arrayBlockingQueue.poll();
    arrayBlockingQueue.poll();
    //取出操作 等待超过2秒就退出
    arrayBlockingQueue.poll(2,TimeUnit.SECONDS);
}

SynchronousQueue同步队列

没有容量,进去一个元素，必须等待取出来之后，才能往里面放一个元素
存：put()
取：take()

public class MySynchronousQueue
{
    public static void main(String[] args)
    {
        //同步队列  synchronousQueue 不存储元素 put 了一个元素必须要从里面取出来 否则不能再put
        SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();


        new Thread(()->{
            for (int i = 0; i <3 ; i++)
            {
                System.out.println(Thread.currentThread().getName()+"put 1");
                try
                {
                    synchronousQueue.put("Y1");
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }


        },"Y1").start();


        new Thread(()->{
            for (int i = 0; i <3; i++)
            {
                try
                {
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName() + synchronousQueue.take());
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        },"Y2").start();
    }
}