BlockingQueue-阿里云开发者社区

开发者社区> 开发与运维> 正文
登录阅读全文

BlockingQueue

简介:  在新增的Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了BlockingQueue家庭中的所有成员,包括他们各自的功能以及常见使用场景。 认识BlockingQueue阻塞队列,顾名思义,首先它是一个队列,而一个队列在数据结构中

 在新增的Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了BlockingQueue家庭中的所有成员,包括他们各自的功能以及常见使用场景。

  • 认识BlockingQueue
    阻塞队列,顾名思义,首先它是一个队列,而一个队列在数据结构中所起的作用大致如下图所示:

    从上图我们可以很清楚看到,通过一个共享的队列,可以使得数据由队列的一端输入,从另外一端输出;
    常用的队列主要有以下两种:(当然通过不同的实现方式,还可以延伸出很多不同类型的队列,DelayQueue就是其中的一种)
      先进先出(FIFO):先插入的队列的元素也最先出队列,类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。
      后进先出(LIFO):后插入队列的元素最先出队列,这种队列优先处理最近发生的事件。

          多线程环境中,通过队列可以很容易实现数据共享,比如经典的“生产者”和“消费者”模型中,通过队列可以很便利地实现两者之间的数据共享。假设我们有若干生产者线程,另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程,利用队列的方式来传递数据,就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内,万一发生数据处理速度不匹配的情况呢?理想情况下,如果生产者产出数据的速度大于消费者消费的速度,并且当生产出来的数据累积到一定程度的时候,那么生产者必须暂停等待一下(阻塞生产者线程),以便等待消费者线程把累积的数据处理完毕,反之亦然。然而,在concurrent包发布以前,在多线程环境下,我们每个程序员都必须去自己控制这些细节,尤其还要兼顾效率和线程安全,而这会给我们的程序带来不小的复杂度。好在此时,强大的concurrent包横空出世了,而他也给我们带来了强大的BlockingQueue。(在多线程领域:所谓阻塞,在某些情况下会挂起线程(即阻塞),一旦条件满足,被挂起的线程又会自动被唤醒)
    下面两幅图演示了BlockingQueue的两个常见阻塞场景:
           如上图所示:当队列中没有数据的情况下,消费者端的所有线程都会被自动阻塞(挂起),直到有数据放入队列。


       如上图所示:当队列中填满数据的情况下,生产者端的所有线程都会被自动阻塞(挂起),直到队列中有空的位置,线程被自动唤醒。
         这也是我们在多线程环境下,为什么需要BlockingQueue的原因。作为BlockingQueue的使用者,我们再也不需要关心什么时候需要阻塞线程,什么时候需要唤醒线程,因为这一切BlockingQueue都给你一手包办了。既然BlockingQueue如此神通广大,让我们一起来见识下它的常用方法:
    BlockingQueue的核心方法
    放入数据:
      offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,
        则返回true,否则返回false.(本方法不阻塞当前执行方法的线程)
      offer(E o, long timeout, TimeUnit unit),可以设定等待的时间,如果在指定的时间内,还不能往队列中
        加入BlockingQueue,则返回失败。
      put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断
        直到BlockingQueue里面有空间再继续.
    获取数据:
      poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,
        取不到时返回null;
      poll(long timeout, TimeUnit unit):从BlockingQueue取出一个队首的对象,如果在指定时间内,
        队列一旦有数据可取,则立即返回队列中的数据。否则知道时间超时还没有数据可取,返回失败。
      take():取走BlockingQueue里排在首位的对象,BlockingQueue为空,阻断进入等待状态直到
        
    BlockingQueue有新的数据被加入; 
      drainTo():一次性从BlockingQueue获取所有可用的数据对象(还可以指定获取数据的个数), 
        通过该方法,可以提升获取数据效率;不需要多次分批加锁或释放锁。
  • 常见BlockingQueue
    在了解了BlockingQueue的基本功能后,让我们来看看BlockingQueue家庭大致有哪些成员? 
     
  • BlockingQueue成员详细介绍
    1. ArrayBlockingQueue
ArrayBlockingQueue是对BlockingQueue的一个数组实现,它使用一把全局的锁并行对queue的读写操作,同时使用两个Condition阻塞容量为空时的取操作和容量满时的写操作,是一个典型的有界缓存区,固定大小的数组在其保持生产者插入的元素和使用者提取的元素。

二、具体实现
    ArrayBlockingQueue底层定义如下:

 

Java代码
public class ArrayBlockingQueue<E> extends AbstractQueue<E>   
        implements BlockingQueue<E>, java.io.Serializable {   
  
    // 使用循环数组来实现queue,初始时takeIndex和putIndex均为0   
    private final E[] items;   
    private transient int takeIndex;   
    private transient int putIndex;   
    private int count;   
  
    // 用于并发的锁和条件   
   private final ReentrantLock lock;   
    private final Condition notEmpty;   
    private final Condition notFull;   
  
      
    final int inc(int i) {   
        return (++i == items.length)? 0 : i;   
    }   
  
    public ArrayBlockingQueue(int capacity, boolean fair) {   
        if (capacity <= 0)   
            throw new IllegalArgumentException();   
        this.items = (E[]) new Object[capacity];   
        // 分配锁及该锁上的condition   
        lock = new ReentrantLock(fair);   
        notEmpty = lock.newCondition();   
        notFull =  lock.newCondition();   
    }   
  
  ...   
}  
   ArrayBlockingQueue的取操作:

Java代码
public class ArrayBlockingQueue<E> extends AbstractQueue<E>   
        implements BlockingQueue<E>, java.io.Serializable {   
  
    private E extract() {   
        final E[] items = this.items;   
        E x = items[takeIndex];   
        items[takeIndex] = null;   
        takeIndex = inc(takeIndex);   
        --count;   
       // 激发notFull条件   
        notFull.signal();   
        return x;   
    }   
  
       
    public E take() throws InterruptedException {   
        final ReentrantLock lock = this.lock;   
        lock.lockInterruptibly();   
        try {   
            try {   
                  // 等待notEmpty的条件   
                while (count == 0)   
                    notEmpty.await();   
            } catch (InterruptedException ie) {   
                notEmpty.signal(); // propagate to non-interrupted thread   
                throw ie;   
            }   
            E x = extract();   
            return x;   
        } finally {   
            lock.unlock();   
        }   
    }   
  
  ...   
}  
   ArrayBlockingQueue的写操作:

Java代码
public class ArrayBlockingQueue<E> extends AbstractQueue<E>   
        implements BlockingQueue<E>, java.io.Serializable {   
  
    private void insert(E x) {   
        items[putIndex] = x;   
        putIndex = inc(putIndex);   
        ++count;   
        notEmpty.signal();   
    }   
  
    public void put(E o) throws InterruptedException {   
        if (o == null) throw new NullPointerException();   
        final E[] items = this.items;   
        final ReentrantLock lock = this.lock;   
        lock.lockInterruptibly();   
        try {   
            try {   
                  // 等待notFull条件   
           while (count == items.length)   
                    notFull.await();   
            } catch (InterruptedException ie) {   
                notFull.signal(); // propagate to non-interrupted thread   
                throw ie;   
            }   
            insert(o);   
        } finally {   
            lock.unlock();   
        }   
    }   
  
  ...   
}  
    注意:ArrayBlockingQueue在读写操作上都需要锁住整个容器,因此吞吐量与一般的实现是相似的,适合于实现“生产者消费者”模式。


LinkedBlockingQueue

LinkedBlockingQueue实现是线程安全的,实现了FIFO(先进先出)等特性. 是作为生产者消费者的首选,LinkedBlockingQueue 可以指定容量,也可以不指定,不指定的话,默认最大是Integer.MAX_VALUE,其中主要用到put和take方法,put方法在队列满的时候会阻塞直到有队列成员被消费,take方
法在队列空的时候会阻塞,直到有队列成员被放进来。

  1. import java.util.concurrent.BlockingQueue;  
  2. import java.util.concurrent.ExecutorService;  
  3. import java.util.concurrent.Executors;  
  4. import java.util.concurrent.LinkedBlockingQueue;  
  5.   
  6. public class MyBlockingQueue extends Thread{  
  7.     public static BlockingQueue<String> queue=new LinkedBlockingQueue<String>(3);  
  8.     private int index;  
  9.     public MyBlockingQueue(int i){  
  10.         this.index=i;  
  11.     }  
  12.   
  13.     public void run(){  
  14.         try{  
  15.             queue.put(String.valueOf(this.index));  
  16.             System.out.println("put {"+this.index+"} into queue!");  
  17.         }catch(Exception e){  
  18.             e.printStackTrace();  
  19.         }  
  20.     }  
  21.       
  22.     public static void main(String args[]){  
  23.         ExecutorService service=Executors.newCachedThreadPool();  
  24.         forint i=0; i<10; i++){  
  25.             service.submit(new MyBlockingQueue(i));  
  26.         }  
  27.         Thread thread = new Thread(){  
  28.             public void run(){  
  29.                 try{  
  30.                     while(true){  
  31.                         Thread.sleep((int)(Math.random()*1000));  
  32.                         if(MyBlockingQueue.queue.isEmpty()) break;  
  33.                         String str=MyBlockingQueue.queue.take();  
  34.                         System.out.println("take {" + str+"} out of queue!");  
  35.                     }  
  36.                 }catch(Exception e){  
  37.                     e.printStackTrace();  
  38.                 }  
  39.             }  
  40.         };  
  41.         service.submit(thread);  
  42.         service.shutdown();  
  43.     }  
  44.       
  45. }         


DelayQueue
      DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据的操作(消费者)才会被阻塞。
使用场景:
  DelayQueue使用场景较少,但都相当巧妙,常见的例子比如使用一个DelayQueue来管理一个超时未响应的连接队列。
PriorityBlockingQueue
      基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定),但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者,而只会在没有可消费的数据时,阻塞数据的消费者。因此使用的时候要特别注意,生产者生产数据的速度绝对不能快于消费者消费数据的速度,否则时间一长,会最终耗尽所有的可用堆内存空间。在实现PriorityBlockingQueue时,内部控制线程同步的锁采用的是公平锁。


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
开发与运维
使用钉钉扫一扫加入圈子
+ 订阅

集结各类场景实战经验,助你开发运维畅行无忧

其他文章
最新文章
相关文章