隊列與阻塞

隊列與阻塞

阻塞隊列：

BlokingQueue

API

主要的Queue的實現(xiàn)類：

AbstractQueue：非阻塞隊列
BlockingQueue： 阻塞隊列
Deque： 雙端隊列

看一下智慧樹

智慧樹

什么情況下我們要使用阻塞隊列？

多線程并發(fā)，線程池

BlockingQueue 四組API

拋出異常：

/*
拋出異常
 */
 public static  void testThrow()
 {
     //   初始化必須 設(shè)置隊列的大小
     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");
    //一直阻塞,一直等待進(jìn)入
    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同步隊列

沒有容量,進(jìn)去一個元素，必須等待取出來之后 ，才能往里面放一個元素
存：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();
    }
}