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

发布时间 - 2026-01-11 02:03:59    点击率:

阻塞队列 (BlockingQueue)是Java util.concurrent包下重要的数据结构,BlockingQueue提供了线程安全的队列访问方式:当阻塞队列进行插入数据时,如果队列已满,线程将会阻塞等待直到队列非满;从阻塞队列取数据时,如果队列已空,线程将会阻塞等待直到队列非空。并发包下很多高级同步类的实现都是基于BlockingQueue实现的。

BlockingQueue 的操作方法

BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到立即执行的话,每个方法的表现也不同。这些方法如下:

四组不同的行为方式解释:

  1. 抛异常:如果试图的操作无法立即执行,抛一个异常。
  2. 特定值:如果试图的操作无法立即执行,返回一个特定的值(常常是 true / false)。
  3. 阻塞:如果试图的操作无法立即执行,该方法调用将会发生阻塞,直到能够执行。
  4. 超时:如果试图的操作无法立即执行,该方法调用将会发生阻塞,直到能够执行,但等待时间不会超过给定值。返回一个特定值以告知该操作是否成功(典型的是true / false)。

无法向一个 BlockingQueue 中插入 null。如果你试图插入 null,BlockingQueue 将会抛出一个 NullPointerException。

可以访问到 BlockingQueue 中的所有元素,而不仅仅是开始和结束的元素。比如说,你将一个对象放入队列之中以等待处理,但你的应用想要将其取消掉。那么你可以调用诸如 remove(o) 方法来将队列之中的特定对象进行移除。但是这么干效率并不高(译者注:基于队列的数据结构,获取除开始或结束位置的其他对象的效率不会太高),因此你尽量不要用这一类的方法,除非你确实不得不那么做。

BlockingQueue 的实现类

BlockingQueue 是个接口,你需要使用它的实现之一来使用BlockingQueue,Java.util.concurrent包下具有以下 BlockingQueue 接口的实现类:

  1. ArrayBlockingQueue:ArrayBlockingQueue 是一个有界的阻塞队列,其内部实现是将对象放到一个数组里。有界也就意味着,它不能够存储无限多数量的元素。它有一个同一时间能够存储元素数量的上限。你可以在对其初始化的时候设定这个上限,但之后就无法对这个上限进行修改了(译者注:因为它是基于数组实现的,也就具有数组的特性:一旦初始化,大小就无法修改)。
  2. DelayQueue:DelayQueue 对元素进行持有直到一个特定的延迟到期。注入其中的元素必须实现 java.util.concurrent.Delayed 接口。
  3. LinkedBlockingQueue:LinkedBlockingQueue 内部以一个链式结构(链接节点)对其元素进行存储。如果需要的话,这一链式结构可以选择一个上限。如果没有定义上限,将使用 Integer.MAX_VALUE 作为上限。
  4. PriorityBlockingQueue:PriorityBlockingQueue 是一个无界的并发队列。它使用了和类 java.util.PriorityQueue 一样的排序规则。你无法向这个队列中插入 null 值。所有插入到 PriorityBlockingQueue 的元素必须实现 java.lang.Comparable 接口。因此该队列中元素的排序就取决于你自己的 Comparable 实现。
  5. SynchronousQueue:SynchronousQueue 是一个特殊的队列,它的内部同时只能够容纳单个元素。如果该队列已有一元素的话,试图向队列中插入一个新元素的线程将会阻塞,直到另一个线程将该元素从队列中抽走。同样,如果该队列为空,试图向队列中抽取一个元素的线程将会阻塞,直到另一个线程向队列中插入了一条新的元素。据此,把这个类称作一个队列显然是夸大其词了。它更多像是一个汇合点。

使用例子:

阻塞队列的最长使用的例子就是生产者消费者模式,也是各种实现生产者消费者模式方式中首选的方式。使用者不用关心什么阻塞生产,什么时候阻塞消费,使用非常方便,代码如下:

package MyThread;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class BlockingQueueTest {
  //生产者
  public static class Producer implements Runnable{
    private final BlockingQueue<Integer> blockingQueue;
    private volatile boolean flag;
    private Random random;

    public Producer(BlockingQueue<Integer> blockingQueue) {
      this.blockingQueue = blockingQueue;
      flag=false;
      random=new Random();

    }
    public void run() {
      while(!flag){
        int info=random.nextInt(100);
        try {
          blockingQueue.put(info);
          System.out.println(Thread.currentThread().getName()+" produce "+info);
          Thread.sleep(50);
        } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }        
      }
    }
    public void shutDown(){
      flag=true;
    }
  }
  //消费者
  public static class Consumer implements Runnable{
    private final BlockingQueue<Integer> blockingQueue;
    private volatile boolean flag;
    public Consumer(BlockingQueue<Integer> blockingQueue) {
      this.blockingQueue = blockingQueue;
    }
    public void run() {
      while(!flag){
        int info;
        try {
          info = blockingQueue.take();
          System.out.println(Thread.currentThread().getName()+" consumer "+info);
          Thread.sleep(50);
        } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }        
      }
    }
    public void shutDown(){
      flag=true;
    }
  }
  public static void main(String[] args){
    BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>(10);
    Producer producer=new Producer(blockingQueue);
    Consumer consumer=new Consumer(blockingQueue);
    //创建5个生产者,5个消费者
    for(int i=0;i<10;i++){
      if(i<5){
        new Thread(producer,"producer"+i).start();
      }else{
        new Thread(consumer,"consumer"+(i-5)).start();
      }
    }

    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    producer.shutDown();
    consumer.shutDown();

  }
}

阻塞队列原理:

其实阻塞队列实现阻塞同步的方式很简单,使用的就是是lock锁的多条件(condition)阻塞控制。使用BlockingQueue封装了根据条件阻塞线程的过程,而我们就不用关心繁琐的await/signal操作了。

下面是Jdk 1.7中ArrayBlockingQueue部分代码:

public ArrayBlockingQueue(int capacity, boolean fair) {

    if (capacity <= 0)
      throw new IllegalArgumentException();
    //创建数组  
    this.items = new Object[capacity];
    //创建锁和阻塞条件
    lock = new ReentrantLock(fair);  
    notEmpty = lock.newCondition();
    notFull = lock.newCondition();
  }
//添加元素的方法
public void put(E e) throws InterruptedException {
    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;
    count++;
    notEmpty.signal();
  }
 //移除元素的方法
 public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
      while (count == 0)
        notEmpty.await();
      return dequeue();
    } finally {
      lock.unlock();
    }
  }
 //出队的方法
 private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
      takeIndex = 0;
    count--;
    if (itrs != null)
      itrs.elementDequeued();
    notFull.signal();
    return x;

双端阻塞队列(BlockingDeque)

concurrent包下还提供双端阻塞队列(BlockingDeque),和BlockingQueue是类似的,只不过BlockingDeque提供从任意一端插入或者抽取元素的队列。

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


# Java阻塞队列  # java阻塞队列原理  # java实现阻塞队列  # 剖析Java中阻塞队列的实现原理及应用场景  # Java 阻塞队列详解及简单使用  # 详解Java中的阻塞队列  # Java并发编程之阻塞队列详解  # Java中使用阻塞队列控制线程集实例  # java阻塞队列实现原理及实例解析  # JavaSE多线程阻塞队列实现代码  # 将会  # 是一个  # 链式  # 移除  # 你可以  # 也就  # 数据结构  # 对其  # 定值  # 自己的  # 的是  # 都是  # 这一  # 是个  # 如果你  # 夸大其词  # 什么时候  # 已有  # 将其  # 而不 


相关栏目: 【 网站优化151355 】 【 网络推广146373 】 【 网络技术251813 】 【 AI营销90571


相关推荐: Laravel路由Route怎么设置_Laravel基础路由定义与参数传递规则【详解】  Laravel如何优雅地处理服务层_在Laravel中使用Service层和Repository层  做企业网站制作流程,企业网站制作基本流程有哪些?  Linux系统命令中tree命令详解  在centOS 7安装mysql 5.7的详细教程  java中使用zxing批量生成二维码立牌  nginx修改上传文件大小限制的方法  html5怎么画眼睛_HT5用Canvas或SVG画眼球瞳孔加JS控制动态【绘制】  zabbix利用python脚本发送报警邮件的方法  Laravel如何创建和注册中间件_Laravel中间件编写与应用流程  微博html5版本怎么弄发超话_超话进入入口及发帖格式要求【教程】  如何在橙子建站中快速调整背景颜色?  java获取注册ip实例  微信小程序制作网站有哪些,微信小程序需要做网站吗?  ,网页ppt怎么弄成自己的ppt?  详解Android——蓝牙技术 带你实现终端间数据传输  如何确认建站备案号应放置的具体位置?  Laravel怎么使用Markdown渲染文档_Laravel将Markdown内容转HTML页面展示【实战】  Laravel如何与Docker(Sail)协同开发?(环境搭建教程)  Windows家庭版如何开启组策略(gpedit.msc)?(安装方法)  网站视频制作书签怎么做,ie浏览器怎么将网站固定在书签工具栏?  如何基于云服务器快速搭建网站及云盘系统?  jquery插件bootstrapValidator表单验证详解  深圳网站制作公司好吗,在深圳找工作哪个网站最好啊?  如何在腾讯云服务器快速搭建个人网站?  哪家制作企业网站好,开办像阿里巴巴那样的网络公司和网站要怎么做?  Laravel如何实现本地化和多语言支持_Laravel多语言配置与翻译文件管理  千库网官网入口推荐 千库网设计创意平台入口  怎么用AI帮你为初创公司进行市场定位分析?  Laravel如何使用Eloquent ORM进行数据库操作?(CRUD示例)  七夕网站制作视频,七夕大促活动怎么报名?  湖南网站制作公司,湖南上善若水科技有限公司做什么的?  Laravel全局作用域是什么_Laravel Eloquent Global Scopes应用指南  清除minerd进程的简单方法  如何在建站主机中优化服务器配置?  如何在阿里云虚拟服务器快速搭建网站?  php485函数参数是什么意思_php485各参数详细说明【介绍】  如何快速搭建个人网站并优化SEO?  如何构建满足综合性能需求的优质建站方案?  高性能网站服务器部署指南:稳定运行与安全配置优化方案  Laravel如何处理文件下载请求?(Response示例)  HTML5段落标签p和br怎么选_文本排版常用标签对比【解答】  Laravel怎么实现验证码功能_Laravel集成验证码库防止机器人注册  如何彻底卸载建站之星软件?  学生网站制作软件,一个12岁的学生写小说,应该去什么样的网站?  创业网站制作流程,创业网站可靠吗?  laravel怎么为API路由添加签名中间件保护_laravel API路由签名中间件保护方法  用v-html解决Vue.js渲染中html标签不被解析的问题  Laravel如何使用Telescope进行调试?(安装和使用教程)  html5audio标签播放结束怎么触发事件_onended回调方法【教程】