JDK1.8 BlockingQueue接口说明

Posted 2022-12-09 zhujm320

介绍

BlockingQueue阻塞队列接口，定义了实现规范，线程安全。方便生成者和消费者模型使用，使用BlockingQueue时，开发者不需要考虑线程安全问题，将精力集中到业务实现即可。实际使用中，以生产者和消费者模型为例，分为三种情况: 1. 队列为空，生产者存入数据，消费者阻塞；2. 队列满，生产者阻塞，消费者拉取数据；3.队列非空非满，生产者可以存入数据，消费者也可以拉取数据。

队列实现原理参考

 

接口介绍

入队操作

boolean add(E e);

添加一个元素到队尾，成功时返回true，如果队列满，则抛出异常

boolean offer(E e);

添加一个元素到队尾，成功时返回true，如果队列满，则返回false

void put(E e) throws InterruptedException;

添加一个元素到队尾，成功时返回true，如果队列满，则阻塞，直到队列有空间

出队操作

E take() throws InterruptedException;

取元素，如果队列非空，直接返回队头元素，反正则阻塞直到队列非空才返回队头元素

E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

取元素，如果队列非空，直接返回队头元素，反之则等待指定的时候，如果有元素则返回元素，反之返回null

队列容量

int remainingCapacity();

返回队列空闲的容量

删除操作

boolean remove(Object o);

删除指定的对象

int drainTo(Collection<? super E> c);

将队列中的元素放到给定的集合中，然后删除队列所有元素

int drainTo(Collection<? super E> c, int maxElements);

将指定大小的元素放到给定集合中，然后再在队列中删除

具体使用用例

入队示例

public void testAdd()
        try 
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
            boolean bRet = false;
            bRet = blockingQueue.add("1");
            bRet = blockingQueue.add("2");
            bRet = blockingQueue.add("3");
            bRet = blockingQueue.add("4");
            bRet = blockingQueue.add("5");
            bRet = blockingQueue.add("6");    //队列满,抛出IllegalStateException异常
            System.out.println("bRet: " + bRet);
        catch (Exception e)
            e.printStackTrace();
        
    

public void testOffer()
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
        boolean bRet = false;
        bRet = blockingQueue.offer("1");
        bRet = blockingQueue.offer("2");
        bRet = blockingQueue.offer("3");
        bRet = blockingQueue.offer("4");
        bRet = blockingQueue.offer("5");
        System.out.println("bRet: " + bRet);
        bRet = blockingQueue.offer("6");
        System.out.println("bRet: " + bRet);
    

public void testPut()
        try
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
            blockingQueue.put("1");
            blockingQueue.put("2");
            blockingQueue.put("3");
            blockingQueue.put("4");
            blockingQueue.put("5");
            System.out.println("put 5");
            blockingQueue.put("6");     //队列满，卡住，等待队列为空返回
            System.out.println("put 6");
        catch (Exception e)
            e.printStackTrace();
        
    

出队示例

public void testTake()
        try
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
            blockingQueue.put("1");
            blockingQueue.put("2");
            blockingQueue.put("3");

            String element = blockingQueue.take();
            System.out.println(element);
            element = blockingQueue.take();
            System.out.println(element);
            element = blockingQueue.take();
            System.out.println(element);
            element = blockingQueue.take();  //队列为空，卡住，直到队列有元素后返回
            System.out.println(element);
        catch (Exception e)
            e.printStackTrace();
        
    

/**
     * 通过poll取元素
     */
    public void testPoll()
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
            blockingQueue.offer("1");
            blockingQueue.offer("2");
            blockingQueue.offer("3");
            try
                String element = blockingQueue.poll(10, TimeUnit.MILLISECONDS);
                System.out.println(element);
                element = blockingQueue.poll(10, TimeUnit.MILLISECONDS);
                System.out.println(element);
                element = blockingQueue.poll(10, TimeUnit.MILLISECONDS);
                System.out.println(element);
                element = blockingQueue.poll(10, TimeUnit.MILLISECONDS);  //队列为空，等待10ms, 如果没有元素返回null
                System.out.println(element);
            catch (Exception e)
                e.printStackTrace();
            
    

删除元素示例

/**
     * 删除单个元素
     */
    public void testRemove()
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
        blockingQueue.offer("1");
        blockingQueue.offer("2");
        blockingQueue.offer("3");
        try
            System.out.println("queue.size: " + blockingQueue.size());
            boolean bRemove = blockingQueue.remove("1");
            System.out.println("remove 1 queue.size: " + blockingQueue.size());
            bRemove = blockingQueue.remove("2");
            System.out.println("remove 2 queue.size: " + blockingQueue.size());
            bRemove = blockingQueue.remove("3");
            System.out.println("remove 3 queue.size: " + blockingQueue.size());
        catch (Exception e)
            e.printStackTrace();
        
    

/**
     * 批量删除
     */
    public void testdrainTo()
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
        blockingQueue.offer("1");
        blockingQueue.offer("2");
        blockingQueue.offer("3");
        try
            System.out.println("queue.size: " + blockingQueue.size());
            List<String> list = new ArrayList<>();
            blockingQueue.drainTo(list);
            System.out.println("queue.size: " + blockingQueue.size());
            System.out.println("list.size: " + list.size());
        catch (Exception e)
            e.printStackTrace();