Java并发编程之阻塞队列详解

1、什么是阻塞队列?
  队列是一种数据结构,它有两个基本操作:在队列尾部加入一个元素,从队列头部移除一个元素。阻塞队里与普通的队列的区别在于,普通队列不会对当前线程产生阻塞,在面对类似消费者-生产者模型时,就必须额外的实现同步策略以及线程间唤醒策略。使用阻塞队列,就会对当前线程产生阻塞,当队列是空时,从队列中获取元素的操作将会被阻塞,当队列是满时,往队列里添加元素的操作也会被阻塞。

2、主要的阻塞队列及其方法

java.util.concurrent包下提供主要的几种阻塞队列,主要有以下几个:

1)ArrayBlockingQueue:基于数组实现的阻塞队列,在创建ArrayBlockingQueue对象时必须指定其容量大小,还可以指定访问策略,默认情况下为非公平的,即不保证等待时间最长的线程最优先能够访问队列。
2)、LinkedBlockingQueue:基于链表实现的一个阻塞队列,在创建LinkedBlockingQueue对象时如果不指定容量大小,则默认大小为Integer.MAX_VALUE。
3)、以上2种队列都是先进先出队列,而PriorityBlockingQueue却不是,它会按照元素的优先级对元素进行排序,按照优先级顺序出队,每次出队的元素都是优先级最高的元素。注意,此阻塞队列为无界阻塞队列,即容量没有上限(通过源码就可以知道,它没有容器满的信号标志),前面2种都是有界队列。
4)、DelayQueue:基于PriorityQueue,一种延时阻塞队列,DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue也是一个无界队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据的操作(消费者)才会被阻塞。

阻塞队列包括了非阻塞队列中的大部分方法,还提供另外若干非常有用的方法:

put方法用来向队尾存入元素,如果队列满,则等待;
take方法用来从队首取元素,如果队列为空,则等待;
offer方法用来向队尾存入元素,如果队列满,则等待一定的时间,当时间期限达到时,如果还没有插入成功,则返回false;否则返回true;
poll方法用来从队首取元素,如果队列空,则等待一定的时间,当时间期限达到时,如果取到,则返回null;否则返回取得的元素;

下面看一段代码:

import java.util.concurrent.ArrayBlockingQueue;

/**
 * @author 作者:徐剑  E-mail:anxu_2013@163.com
 * @version 创建时间:2016年3月20日 下午12:52:53
 * 类说明
 */
public class BlockingQueue
{
  public static void main(String[] args) throws InterruptedException
  {
    java.util.concurrent.BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
    for (int i = 0; i < 10; i++)
    {
      // 将指定元素添加到此队列中
      blockingQueue.put("加入元素" + i);
      System.out.println("向阻塞队列中添加了元素:" + i);
    }
    System.out.println("程序到此运行结束,即将退出----");
  }
}

  当限制阻塞队列数量为5时,添加了5个元素之后,继续添加将会队列外阻塞等待,此时程序并未终止。

  

  当队列满了之后,我们将队首元素移除,则可以继续向阻塞队列中添加元素,代码如下:

public class BlockingQueue
{
  public static void main(String[] args) throws InterruptedException
  {
    java.util.concurrent.BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(5);
    for (int i = 0; i < 10; i++)
    {
      // 将指定元素添加到此队列中
      blockingQueue.put("加入元素" + i);
      System.out.println("向阻塞队列中添加了元素:" + i);
      if(i>=4)
        System.out.println("移除队首元素"+blockingQueue.take());
    }
    System.out.println("程序到此运行结束,即将退出----");
  }

  执行结果如下:

  

3、阻塞队列的实现原理
  下面主要看一下ArrayBlockingQueue的实现原理。

  首先看一下ArrayBlockingQueue类的成员变量:

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {

  /** 底层存储结构-数组 */
  final Object[] items;

  /** 队首元素下标 */
  int takeIndex;

  /** 队尾元素下标 */
  int putIndex;

  /**队列元素总数 */
  int count;

  /** 重入锁 */
  final ReentrantLock lock;

  /** notEmpty等待条件 */
  private final Condition notEmpty;
  /** notFull等待条件 */
  private final Condition notFull;
  /**
   * Shared state for currently active iterators, or null if there
   * are known not to be any. Allows queue operations to update
   * iterator state.
   */
  transient Itrs itrs = null;

可以看到,ArrayBlockingQueue用来存储元素的实际上是一个数组。

再看下ArrayBlockingQueue两个重要方法的实现,put()和take():

public void put(E e) throws InterruptedException
  {
    //先检查e是否为空
    checkNotNull(e);
    //获取锁
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try
    {
      //当队列已满,进入条件等待
      while (count == items.length)
        notFull.await();
      //队列不满,进行入队列操作
      enqueue(e);
    }
    finally
    {
      //释放锁
      lock.unlock();
    }
  }

再看下具体的入队操作:

private void enqueue(E x)
  {
    final Object[] items = this.items;
    //队尾入队
    items[putIndex] = x;
    if (++putIndex == items.length)
      putIndex = 0;
    //队列总数+1
    count++;
    //notempty条件的等待集中随机选择一个线程,解除其阻塞状态
    notEmpty.signal();
  }

下面是take()方法的源代码:

public E take() throws InterruptedException
  {
    //获取锁
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try
    {
      //队列为空
      while (count == 0)
        //线程加入notEmpty条件等待集
        notEmpty.await();
      //非空,出队列
      return dequeue();
    } finally
    {
      //释放锁
      lock.unlock();
    }
  }

4、阻塞队列的应用:实现消费者-生产者模式

/**
 * @author 作者:徐剑  E-mail:anxu_2013@163.com
 * @version 创建时间:2016年3月20日 下午2:21:55
 * 类说明:阻塞队列实现的消费者-生产者模式
 */
public class Test
{
  private int queueSize = 10;
  private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(queueSize);
  public static void main(String[] args)
  {
    Test test = new Test();
    Producer producer = test.new Producer();
    Consumer consumer = test.new Consumer();
    producer.start();
    consumer.start();
  }
  class Consumer extends Thread
  {
    @Override
    public void run()
    {
      consume();
    }

    private void consume()
    {
      while (true)
      {
        try
        {
          queue.take();
          System.out.println("从队列取走一个元素,队列剩余" + queue.size() + "个元素");
        } catch (InterruptedException e)
        {
          e.printStackTrace();
        }
      }
    }
  }
  class Producer extends Thread
  {
    @Override
    public void run()
    {
      produce();
    }
    private void produce()
    {
      while (true)
      {
        try
        {
          queue.put(1);
          System.out.println("向队列取中插入一个元素,队列剩余空间:"+ (queueSize - queue.size()));
        } catch (InterruptedException e)
        {
          e.printStackTrace();
        }
      }
    }
  }
}

以上就是本文的全部内容,希望对大家的学习有所帮助。

(0)

相关推荐

  • Java中的阻塞队列详细介绍

    Java中的阻塞队列 1. 什么是阻塞队列? 阻塞队列(BlockingQueue)是一个支持两个附加操作的队列.这两个附加的操作是: 在队列为空时,获取元素的线程会等待队列变为非空. 当队列满时,存储元素的线程会等待队列可用. 阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程.阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素. 2.Java里的阻塞队列 JDK中提供了七个阻塞队列: ArrayBlockingQueue :一个由数组结

  • 深入理解Java线程编程中的阻塞队列容器

    1. 什么是阻塞队列? 阻塞队列(BlockingQueue)是一个支持两个附加操作的队列.这两个附加的操作是:在队列为空时,获取元素的线程会等待队列变为非空.当队列满时,存储元素的线程会等待队列可用.阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程.阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素. 阻塞队列提供了四种处理方法: 抛出异常:是指当阻塞队列满时候,再往队列里插入元素,会抛出IllegalStateException("Q

  • java 中 阻塞队列BlockingQueue详解及实例

    java 中 阻塞队列BlockingQueue详解及实例 BlockingQueue很好的解决了多线程中数据的传输,首先BlockingQueue是一个接口,它大致有四个实现类,这是一个很特殊的队列,如果BlockQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直到BlockingQueue进了东西才会被唤醒.同样,如果BlockingQueue是满的,任何试图往里存东西的操作也会被阻断进入等待状态,直到BlockingQueue里有空间才会被唤醒继续操作.

  • Java 阻塞队列详解及简单使用

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

  • Java中使用阻塞队列控制线程集实例

    队列以一种先进先出的方式管理数据.如果你试图向一个已经满了的阻塞队列中添加一个元素,或是从一个空的阻塞队列中移除一个元素,将导致线程阻塞.在多线程进行合作时,阻塞队列是很有用的工具.工作者线程可以定期的把中间结果存到阻塞队列中.而其他工作者线程把中间结果取出并在将来修改它们.队列会自动平衡负载.如果第一个线程集运行的比第二个慢,则第二个线程集在等待结果时就会阻塞.如果第一个线程集运行的快,那么它将等待第二个线程集赶上来. 下面的程序展示了如何使用阻塞队列来控制线程集.程序在一个目录及它的所有子目

  • 详解Java阻塞队列(BlockingQueue)的实现原理

    阻塞队列 (BlockingQueue)是Java util.concurrent包下重要的数据结构,BlockingQueue提供了线程安全的队列访问方式:当阻塞队列进行插入数据时,如果队列已满,线程将会阻塞等待直到队列非满:从阻塞队列取数据时,如果队列已空,线程将会阻塞等待直到队列非空.并发包下很多高级同步类的实现都是基于BlockingQueue实现的. BlockingQueue 的操作方法 BlockingQueue 具有 4 组不同的方法用于插入.移除以及对队列中的元素进行检查.如果

  • 剖析Java中阻塞队列的实现原理及应用场景

    我们平时使用的一些常见队列都是非阻塞队列,比如PriorityQueue.LinkedList(LinkedList是双向链表,它实现了Dequeue接口). 使用非阻塞队列的时候有一个很大问题就是:它不会对当前线程产生阻塞,那么在面对类似消费者-生产者的模型时,就必须额外地实现同步策略以及线程间唤醒策略,这个实现起来就非常麻烦.但是有了阻塞队列就不一样了,它会对当前线程产生阻塞,比如一个线程从一个空的阻塞队列中取元素,此时线程会被阻塞直到阻塞队列中有了元素.当队列中有元素后,被阻塞的线程会自动

  • Java并发编程之阻塞队列详解

    1.什么是阻塞队列? 队列是一种数据结构,它有两个基本操作:在队列尾部加入一个元素,从队列头部移除一个元素.阻塞队里与普通的队列的区别在于,普通队列不会对当前线程产生阻塞,在面对类似消费者-生产者模型时,就必须额外的实现同步策略以及线程间唤醒策略.使用阻塞队列,就会对当前线程产生阻塞,当队列是空时,从队列中获取元素的操作将会被阻塞,当队列是满时,往队列里添加元素的操作也会被阻塞. 2.主要的阻塞队列及其方法 java.util.concurrent包下提供主要的几种阻塞队列,主要有以下几个: 1

  • Java并发编程之阻塞队列深入详解

    目录 1. 什么是阻塞队列 2. 阻塞队列的代码使用 3. 生产者消费者模型 (1)应用一:解耦合 (2)应用二:削峰填谷 (3)相关代码 4.阻塞队列和生产者消费者模型功能的实现 1. 什么是阻塞队列 阻塞队列是一种特殊的队列,和数据结构中普通的队列一样,也遵守先进先出的原则同时,阻塞队列是一种能保证线程安全的数据结构,并且具有以下两种特性:当队列满的时候,继续向队列中插入元素就会让队列阻塞,直到有其他线程从队列中取走元素:当队列为空的时候,继续出队列也会让队列阻塞,直到有其他线程往队列中插入

  • Java并发编程Semaphore计数信号量详解

    Semaphore 是一个计数信号量,它的本质是一个共享锁.信号量维护了一个信号量许可集.线程可以通过调用acquire()来获取信号量的许可:当信号量中有可用的许可时,线程能获取该许可:否则线程必须等待,直到有可用的许可为止. 线程可以通过release()来释放它所持有的信号量许可(用完信号量之后必须释放,不然其他线程可能会无法获取信号量). 简单示例: package me.socketthread; import java.util.concurrent.ExecutorService;

  • Java并发编程总结——慎用CAS详解

    一.CAS和synchronized适用场景 1.对于资源竞争较少的情况,使用synchronized同步锁进行线程阻塞和唤醒切换以及用户态内核态间的切换操作额外浪费消耗cpu资源:而CAS基于硬件实现,不需要进入内核,不需要切换线程,操作自旋几率较少,因此可以获得更高的性能. 2.对于资源竞争严重的情况,CAS自旋的概率会比较大,从而浪费更多的CPU资源,效率低于synchronized.以java.util.concurrent.atomic包中AtomicInteger类为例,其getAn

  • Java并发编程预防死锁过程详解

    这篇文章主要介绍了Java并发编程预防死锁过程详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 在java并发编程领域已经有技术大咖总结出了发生死锁的条件,只有四个条件都发生时才会出现死锁: 1.互斥,共享资源X和Y只能被一个线程占用 2.占有且等待,线程T1已经取得共享资源X,在等待共享资源Y的时候,不释放共享资源X 3.不可抢占,其他线程不能强行抢占线程T1占有的资源 4.循环等待,线程T1等待线程T2占有的资源,线程T2等待线程T1占有

  • Java多线程案例之阻塞队列详解

    目录 一.阻塞队列介绍 1.1阻塞队列特性 1.2阻塞队列的优点 二.生产者消费者模型 2.1阻塞队列对生产者的优化 三.标准库中的阻塞队列 3.1Java提供阻塞队列实现的标准类 3.2Blockingqueue基本使用 四.阻塞队列实现 4.1阻塞队列的代码实现 4.2阻塞队列搭配生产者与消费者的代码实现 一.阻塞队列介绍 1.1阻塞队列特性 阻塞队列特性: 一.安全性 二.产生阻塞效果 阻塞队列是一种特殊的队列. 也遵守 “先进先出” 的原则.阻塞队列能是一种线程安全的数据结构, 并且具有

  • Java并发编程之阻塞队列(BlockingQueue)详解

    目录 队列 阻塞队列 ArrayBlockingQueue 重要属性 构造方法 添加元素 add(e) offer(e) put(e) offer(e,time,unit) 移除元素 take() dequeue() LinkedBlockingQueue 重要属性 构造方法 添加元素 offer(e) put(e) 移除元素 poll() take() 对比 总结 大家好,我是小黑,一个在互联网苟且偷生的农民工. 队列 学过数据结构的同学应该都知道,队列是数据结构中一种特殊的线性表结构,和平时

  • Java并发编程之Exchanger方法详解

    简介 Exchanger是一个用于线程间数据交换的工具类,它提供一个公共点,在这个公共点,两个线程可以交换彼此的数据. 当一个线程调用exchange方法后将进入等待状态,直到另外一个线程调用exchange方法,双方完成数据交换后继续执行. Exchanger的使用 方法介绍 exchange(V x):阻塞当前线程,直到另外一个线程调用exchange方法或者当前线程被中断. x : 需要交换的对象. exchange(V x, long timeout, TimeUnit unit):阻塞

  • Java并发编程之Executors类详解

    一.Executors的理解 Executors类属于java.util.concurrent包: 线程池的创建分为两种方式:ThreadPoolExecutor 和 Executors: Executors(静态Executor工厂)用于创建线程池: 工厂和工具方法Executor , ExecutorService , ScheduledExecutorService , ThreadFactory和Callable在此包中定义的类: jdk1.8API中的解释如下: 二.Executors

随机推荐