消息队列常用于有生产者和消费者两类角色的多线程同步场景

BlockingQueue也是java.util.concurrent下的主要用来控制线程同步的工具。

主要的方法是：put、take一对阻塞存取；add、poll一对非阻塞存取。

插入:

1)add(anObject):把anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则抛出异常

2)offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.

3)put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻塞直到BlockingQueue里面有空间再继续.

读取：

4)poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null

5)take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止

其他

int remainingCapacity();返回队列剩余的容量，在队列插入和获取的时候，不要瞎搞，数 据可能不准

boolean remove(Object o); 从队列移除元素，如果存在，即移除一个或者更多，队列改    变了返回true

public boolean contains(Object o); 查看队列是否存在这个元素，存在返回true

int drainTo(Collection<? super E> c); 传入的集合中的元素，如果在队列中存在，那么将     队列中的元素移动到集合中

int drainTo(Collection<? super E> c, int maxElements); 和上面方法的区别在于，制定了移   动的数量

案例：

package blockingqueue;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {

BlockingQueue<String> queue;

public Consumer(BlockingQueue<String> queue) {

this.queue = queue;

}

@Override

public void run() {

try {

String consumer = Thread.currentThread().getName();

System.out.println(consumer);

//如果队列为空，会阻塞当前线程

String temp = queue.take();

System.out.println(consumer + "消费者  get a product:" + temp);

} catch (Exception e) {

e.printStackTrace();

}

}

}

package blockingqueue;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

BlockingQueue<String> queue;

public Producer(BlockingQueue<String> queue) {

this.queue = queue;

}

@Override

public void run() {

try {

String temp = "A Product, 生产线程："

+ Thread.currentThread().getName();

queue.put(temp);//如果队列是满的话，会阻塞当前线程

System.out.println("生产者 I have made a product: "

+ Thread.currentThread().getName());

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

package blockingqueue;

 

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

 

public class Test {

 

   public static void main(String[] args) throws Exception {

      BlockingQueue<String> queue = new LinkedBlockingQueue<String>(2);

      // BlockingQueue<String> queue = new LinkedBlockingQueue<String>();

      // 不设置的话，LinkedBlockingQueue默认大小为Integer.MAX_VALUE

      // BlockingQueue<String> queue = new ArrayBlockingQueue<String>(2);

      Consumer consumer = new Consumer(queue);

      Producer producer = new Producer(queue);

      for (int i = 0; i < 3; i++) {

         new Thread(producer, "Producer" + (i + 1)).start();

      }

      for (int i = 0; i < 5; i++) {

         new Thread(consumer, "Consumer" + (i + 1)).start();

      }

     

      Thread.sleep(5000);

     

      new Thread(producer, "Producer" + (5)).start();

   }

}

BlockingQueue有四个具体的实现类,常用的两种实现类为：

1、ArrayBlockingQueue：一个由数组支持的有界阻塞队列，规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小.其所含的对象是以FIFO(先入先出)顺序排序的。

2、LinkedBlockingQueue：大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的。

LinkedBlockingQueue 可以指定容量，也可以不指定，不指定的话，默认最大是Integer.MAX_VALUE,其中主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。

LinkedBlockingQueue和ArrayBlockingQueue区别：

LinkedBlockingQueue和ArrayBlockingQueue比较起来,它们背后所用的数据结构不一样,导致LinkedBlockingQueue的数据吞吐量要大于ArrayBlockingQueue,但在线程数量很大时其性能的可预见性低于ArrayBlockingQueue.

生产者消费者的示例代码：

见代码  TestBlockingQueue  TestBlockingQueueConsumer   TestBlockingQueueProducer

package blockingqueue;

import java.util.Random;

import java.util.concurrent.BlockingQueue;

public class TestBlockingQueueProducer implements Runnable {

BlockingQueue<String> queue;

Random random = new Random();

public TestBlockingQueueProducer(BlockingQueue<String> queue) {

this.queue = queue;

}

@Override

public void run() {

for (int i = 0; i < 10; i++) {

try {

Thread.sleep(random.nextInt(10));

String task = Thread.currentThread().getName() + " made a product " + i;

System.out.println(task);

queue.put(task);  //阻塞方法

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

package blockingqueue;

 

import java.util.Random;

import java.util.concurrent.BlockingQueue;

 

public class TestBlockingQueueConsumer implements Runnable {

   BlockingQueue<String> queue;

    Random random = new Random();

   

    public TestBlockingQueueConsumer(BlockingQueue<String> queue){ 

        this.queue = queue; 

    }       

    @Override 

    public void run() { 

        try { 

          Thread.sleep(random.nextInt(10));

          System.out.println(Thread.currentThread().getName()+ "trying...");

            String temp = queue.take();//如果队列为空，会阻塞当前线程 

            int remainingCapacity = queue.remainingCapacity();

            System.out.println(Thread.currentThread().getName() + " get a job " +temp);

            // System.out.println("队列中的元素个数： "+ remainingCapacity);

        } catch (InterruptedException e) { 

            e.printStackTrace();

        } 

    }

}

package blockingqueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

public class TestBlockingQueue {

public static void main(String[] args) {

BlockingQueue<String> queue = new LinkedBlockingQueue<String>(2);

// BlockingQueue<String> queue = new LinkedBlockingQueue<String>();

// 不设置的话，LinkedBlockingQueue默认大小为Integer.MAX_VALUE

// BlockingQueue<String> queue = new ArrayBlockingQueue<String>(2);

TestBlockingQueueConsumer consumer = new TestBlockingQueueConsumer(queue);

TestBlockingQueueProducer producer = new TestBlockingQueueProducer(queue);

for (int i = 0; i < 3; i++) {

new Thread(producer, "Producer" + (i + 1)).start();

}

for (int i = 0; i < 5; i++) {

new Thread(consumer, "Consumer" + (i + 1)).start();

}

}

}

java并发包消息队列相关推荐

  1. Java 帝国之消息队列

    张家村的历史 Java 帝国的张家村正在迎来一次重大的变革. 5年前网上购物兴起的时候, 帝国非常看好, 决定向这个领域进军, 于是兴建了张家村, 在这里安装了Java 虚拟机和数据库, 然后部署了一 ...

  2. rabbitmq java实例_RabbitMQ消息队列入门篇（环境配置+Java实例+基础概念）

    转载http://blog.csdn.net/u013142781 一.消息队列使用场景或者其好处 消息队列一般是在项目中,将一些无需即时返回且耗时的操作提取出来,进行了异步处理,而这种异步处理的方式 ...

  3. activimq java集成_Java消息队列-Spring整合ActiveMq

    1.概述 首先和大家一起回顾一下Java 消息服务,在我之前的博客<Java消息队列-JMS概述>中,我为大家分析了: 消息服务:一个中间件,用于解决两个活多个程序之间的耦合,底层由Jav ...

  4. java中的消息队列

    消息队列:可以看做是一个存储消息的容器,它是分布式系统中的重要组件之一. 目的是: 1.为了通过异步处理来提高系统的性能来减少系统响应的时间 一般的步骤是客户端发起请求给服务端,服务端在请求给数据库, ...

  5. Java redis实现消息队列

    文章目录 一.单元测试Java多线程 二.redis实现消息队列 三.java多线程模拟生产者消费者 四.阻塞读 一.单元测试Java多线程 使用junit测试多线程代码,但是等到程序结束,输出结果不 ...

  6. Java实现redis消息队列发布/订阅模式

    最近在一个老项目中需要用消息队列,本来想着用卡夫卡,但是试了几个版本之后发现jdk和卡夫卡版本一直对不上,最后选择用redis来实现消息队列的发布/订阅模式.感谢这位大佬的博客给了我很多的帮助,htt ...

  7. Java笔试面试-消息队列面试题总结

    1.消息队列的应用场景有哪些? 答:消息队列的应用场景如下. 应用解耦,比如,用户下单后,订单系统需要通知库存系统,假如库存系统无法访问,则订单减库存将失败,从而导致订单失败.订单系统与库存系统耦合, ...

  8. Java架构之消息队列 （一）：消息队列的概述

    消息队列系列分享大纲: 一.消息队列的概述 二.消息队列之RabbitMQ的使用 三.消息队列之Kafka的使用 四.消息队列之RabbitMQ的原理详解 五.消息队列之Kafka的原理详解 六.消息 ...

  9. 消息队列如何使用java,想使用消息队列，先考虑下这些问题！，消息队列如何使用...

    想使用消息队列,先考虑下这些问题!,消息队列如何使用原创:Java派(微信公众号:Java派),欢迎分享,转载请保留出处. 消息队列优势 消息队列(Message Queue,简称MQ),其主要用于在 ...

最新文章

  1. 23.使用load()方法异步请求数据
  2. linux系统启动过程（二）
  3. DL之CNN：卷积神经网络算法简介之原理简介——CNN网络的3D可视化(LeNet-5为例可视化)
  4. ASP.NET Core开发-使用Nancy框架
  5. 每天学点Python之collections
  6. 源路由 小实验 小发现
  7. 【渝粤教育】国家开放大学2018年春季 8634-22TAndroid智能手机编程 参考试题
  8. 性能服务器如何放进ppt,Linux服务器性能评估与优化.ppt
  9. MFC使用简单总结（便于以后查阅）
  10. 洛谷3389：【模板】高斯消元法——题解
  11. 使用UUID和int自增主键的区别
  12. paip. 定时 关机 休眠 的总结
  13. Nginx：屏蔽 IP
  14. 查看本地ip地址和端口方法
  15. 第一型曲线和曲面积分总结
  16. 一个屌丝程序员的青春（三二三）
  17. cocos2dx android obb,cocos2dx 实现obb包读取 quick2.2.6
  18. LDAP简述及统一认证说明
  19. linux leanote云笔记搭建
  20. 小米与美的的“初吻”是为了什么

热门文章

  1. 动态规划原理介绍（附7个算例，有代码讲解）
  2. MyBatis全局配置文件介绍
  3. JavaScript实现找出一个数的质因数primeFactors算法（附完整源码）
  4. JavaScript实现按位的isPowerOfTwo算法（附完整源码）
  5. wxWidgets：国际化
  6. boost::intrusive_ref_counter相关的测试程序
  7. boost::mpl模块实现empty_sequence相关的测试程序
  8. boost::mp11::mp_count相关用法的测试程序
  9. boost::hana::permutations用法的测试程序
  10. bgi::detail::is_valid用法的测试程序