文档章节

Java并发(四)BlockingQueue的使用

摆渡者
 摆渡者
发布于 2015/10/12 10:57
字数 1222
阅读 547
收藏 5
点赞 0
评论 0

    wait()和notifyAll()方法以一种非常低级的方式解决了任务互操作的问题,即每次交互时都需要握手。在许多情况下,你可以瞄准更高的抽象级别,使用同步队列来解决任务协作的问题。同步队列在任何时刻都只允许一个任务插入或移除元素。在java.util.concurrent.BlockingQueue接口中提供了这个队列,这个接口有大量的标准实现。你通常可以使用LinkedBlockingQueue,它是一个无届队列,你还可以使用ArrayBlockingQueue,它具有固定的尺寸,因此你可以在它被阻塞之前,向其中放置有限数量的元素。

    如果消费者任务试图从队列中获取对象,而该队列此时为空,那么这些队列还可以挂起消费者任务,并且当有更多的元素可用时回复消费者任务。阻塞队列可以解决非常大的问题,而其方式与wait()和notifyAll()相比,则要简单并可靠许多。

    考虑下面这个BlockingQueue的示例,有一台机器具有三个任务:一个制作吐司,一个给吐司抹黄油,还有一个给吐司涂果酱。我们可以通过各个处理过程之间的BlockingQueue来运行这个吐司制作程序:

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

class Toast {
    /**
     * 吐司的状态:
     * DRY: 烘干的
     * BUTTERED: 涂了黄油的
     * JAMMED: 涂了果酱的
     * <p>吐司的状态只能由DRY->BUTTERED->JAMMED转变
     */
    public enum Status {DRY, BUTTERED, JAMMED}
    private Status status = Status.DRY;//默认状态为DRY
    private final int id;
    public Toast(int id) { this.id =  id;}
    public void butter() {status = Status.BUTTERED;}
    public void jam() {status = Status.JAMMED;}
    public Status getStatus() {return status;}
    public int getId() {return id;}
    public String toString() {
        return "Toast id: " + id + ", status: " + status;
    }
}

@SuppressWarnings("serial")
class ToastQueue extends LinkedBlockingQueue<Toast> {}

/**
 * 生产吐司的任务。
 */
class Toaster implements Runnable {
    private ToastQueue toastQueue;
    private int count = 0;
    private Random random = new Random(47);
    public Toaster(ToastQueue queue) {
        this.toastQueue = queue;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                TimeUnit.MILLISECONDS.sleep(300 + random.nextInt(500));
                //生产一片吐司,这些吐司是有序的
                Toast toast = new Toast(count++);
                System.out.println(toast);
                //放到toastQueue中
                toastQueue.put(toast);
            }
        } catch (InterruptedException e) {
            System.out.println("Toaster interrupted.");
        }
        System.out.println("Toaster off.");
    }
}

/**
 * 涂黄油的任务。
 */
class Butterer implements Runnable {
    private ToastQueue dryQueue;
    private ToastQueue butteredQueue;
    public Butterer(ToastQueue dryQueue, ToastQueue butteredQueue) {
        this.dryQueue = dryQueue;
        this.butteredQueue = butteredQueue;
    }
    
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                //在取得下一个吐司之前会一直阻塞
                Toast toast = dryQueue.take();
                toast.butter();
                System.out.println(toast);
                butteredQueue.put(toast);
            }
        } catch (InterruptedException e) {
            System.out.println("Butterer interrupted.");
        }
        System.out.println("Butterer off.");
        
    }
}

/**
 * 涂果酱的任务。
 */
class Jammer implements Runnable {
    private ToastQueue butteredQueue;
    private ToastQueue finishedQueue;
    public Jammer(ToastQueue butteredQueue, ToastQueue finishedQueue) {
        this.finishedQueue = finishedQueue;
        this.butteredQueue = butteredQueue;
    }
    
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                //在取得下一个吐司之前会一直阻塞
                Toast toast = butteredQueue.take();
                toast.jam();
                System.out.println(toast);
                finishedQueue.put(toast);
            }
        } catch (InterruptedException e) {
            System.out.println("Jammer interrupted.");
        }
        System.out.println("Jammer off.");
    }
}

/**
 * 吃吐司的人,消费者。
 */
class Eater implements Runnable {
    private ToastQueue finishedQueue;
    private int count = 0;
    public Eater (ToastQueue finishedQueue) {
        this.finishedQueue = finishedQueue;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                //在取得下一个吐司之前会一直阻塞
                Toast toast = finishedQueue.take();
                //验证取得的吐司是有序的,而且状态是JAMMED的
                if (toast.getId() != count++ || 
                        toast.getStatus() != Toast.Status.JAMMED) {
                    System.out.println("Error -> " + toast);
                    System.exit(-1);
                } else {
                    //吃掉吐司
                    System.out.println(toast + "->Eaten");
                }
            }
        } catch (InterruptedException e) {
            System.out.println("Eater interrupted.");
        }
        System.out.println("Eater off.");
    }
}

public class ToastOMatic {
    public static void main(String[] args) throws Exception {
        ToastQueue dryQueue = new ToastQueue();
        ToastQueue butteredQueue = new ToastQueue();
        ToastQueue finishedQueue = new ToastQueue();
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new Toaster(dryQueue));
        exec.execute(new Butterer(dryQueue, butteredQueue));
        exec.execute(new Jammer(butteredQueue, finishedQueue));
        exec.execute(new Eater(finishedQueue));
        TimeUnit.SECONDS.sleep(5);
        exec.shutdownNow();
    }
}

执行结果(可能的结果):

Toast id: 0, status: DRY
Toast id: 0, status: BUTTERED
Toast id: 0, status: JAMMED
Toast id: 0, status: JAMMED->Eaten
Toast id: 1, status: DRY
Toast id: 1, status: BUTTERED
Toast id: 1, status: JAMMED
Toast id: 1, status: JAMMED->Eaten
Toast id: 2, status: DRY
Toast id: 2, status: BUTTERED
Toast id: 2, status: JAMMED
Toast id: 2, status: JAMMED->Eaten
Toast id: 3, status: DRY
Toast id: 3, status: BUTTERED
Toast id: 3, status: JAMMED
Toast id: 3, status: JAMMED->Eaten
Toast id: 4, status: DRY
Toast id: 4, status: BUTTERED
Toast id: 4, status: JAMMED
Toast id: 4, status: JAMMED->Eaten
Toast id: 5, status: DRY
Toast id: 5, status: BUTTERED
Toast id: 5, status: JAMMED
Toast id: 5, status: JAMMED->Eaten
Toast id: 6, status: DRY
Toast id: 6, status: BUTTERED
Toast id: 6, status: JAMMED
Toast id: 6, status: JAMMED->Eaten
Toast id: 7, status: DRY
Toast id: 7, status: BUTTERED
Toast id: 7, status: JAMMED
Toast id: 7, status: JAMMED->Eaten
Eater interrupted.
Eater off.
Butterer interrupted.
Toaster interrupted.
Toaster off.
Jammer interrupted.
Jammer off.
Butterer off.

    Toast是一个使用enum值的优秀示例。注意,这个示例中没有任何显式的同步(即使用Lock对象或者synchronized关键字的同步),因为同步已经由队列和系统的设计隐式的管理了——每片Toast在任何时刻都只由一个任务在操作。因为队列的阻塞,使得处理过程将被自动的挂起和恢复。你可以看到由BlockingQueue产生的简化十分明显。在使用显式的wait()和notifyAll()时存在的类和类之间的耦合被消除了,因为每个类都只和它的BlockingQueue通信。

© 著作权归作者所有

共有 人打赏支持
摆渡者
粉丝 318
博文 169
码字总数 205794
作品 0
浦东
程序员
【Java并发专题】27篇文章详细总结Java并发基础知识

努力的意义,就是,在以后的日子里,放眼望去全是自己喜欢的人和事! github:https://github.com/CL0610/Java-concurrency,欢迎题issue和Pull request。所有的文档都是自己亲自码的,如果觉...

你听___
05/06
0
0
Java并发编程利用 Condition 实现阻塞队列

什么是阻塞队列 BlockingQueue 队列是一种数据结构,它的特点是先进先出(First In First Out),它有两个基本操作:在队列尾部加入一个元素,从队列头部移除一个元素。队列在多线程应用中,...

行走在旅途中
2017/11/07
0
0
Java 并发工具包 java.util.concurrent 用户指南

译序 本指南根据 Jakob Jenkov 最新博客翻译,请随时关注博客更新:http://tutorials.jenkov.com/java-util-concurrent/index.html。 本指南已做成中英文对照阅读版的 pdf 文档,有兴趣的朋友...

pior
2015/10/26
0
0
Java 使用 happen-before 规则实现共享变量的同步操作

前言 熟悉 Java 并发编程的都知道,JMM(Java 内存模型) 中的 happen-before(简称 hb)规则,该规则定义了 Java 多线程操作的有序性和可见性,防止了编译器重排序对程序结果的影响。按照官方的...

stateIs0
01/20
0
0
并发队列ConcurrentLinkedQueue和阻塞队列LinkedBlockingQueue用

在Java多线程应用中,队列的使用率很高,多数生产消费模型的首选数据结构就是队列。Java提供的线程安全的Queue可以分为阻塞队列和非阻塞队列,其中阻塞队列的典型例子是BlockingQueue,非阻塞...

凯文加内特
2014/07/31
0
0
BlockingQueue-线程的阻塞队列

BlockingQueue作为线程容器,可以为线程同步提供有力的保障,其主要用到的方法包括: [java] add(E o); //将指定的元素添加到此队列中(如果立即可行),在成功时返回 true,其他情况则抛出 ...

方绍伟
2013/09/24
0
0
并发队列ConcurrentLinkedQueue和阻塞队列LinkedBlockingQueue用法

在Java多线程应用中,队列的使用率很高,多数生产消费模型的首选数据结构就是队列。Java提供的线程安全的Queue可以分为阻塞队列和非阻塞队列,其中阻塞队列的典型例子是BlockingQueue,非阻塞...

F风向标F
2013/12/03
0
0
Java Tomcat vs PHP CLI Server 性能对比

测试环境: Ubuntu 14.04 with i5-3230M Tomcat 8.0.5 with 64bit jre1.7.0_55 PHP 5.5.8 with Zend OPcache v7.0.3-dev 测试脚本: 显示当前服务器时间 Java Tomcat: t.jsp <%@ page session=......

eechen
2014/09/22
0
34
Java:ConcurrentHashMap的实现机制

探索 ConcurrentHashMap 高并发性的实现机制 Java并发编程之ConcurrentHashMap 聊聊并发(四)——深入分析ConcurrentHashMap 上面的三篇分析是针对java7的,java8中的实现方式已经变化。...

樂天
2015/06/28
0
0
深入研究java.lang.ThreadLocal类

一、概述 ThreadLocal是什么呢?其实ThreadLocal并非是一个线程的本地实现版本,它并不是一个Thread,而是threadlocalvariable(线程局部变量)。也许把它命名为ThreadLocalVar更加合适。线程局...

FoxHu
2012/05/08
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

sklearn学习笔记之简单线性回归

简单线性回归 线性回归是数据挖掘中的基础算法之一,从某种意义上来说,在学习函数的时候已经开始接触线性回归了,只不过那时候并没有涉及到误差项。线性回归的思想其实就是解一组方程,得到...

wangxuwei
9分钟前
0
0
feign之动态interceptor(二)

背景 上文提到了按照不同的feignClient可以根据多个不同的key来进行多个不同的bean的配置 那么我们如何完成多个interceptor的配置呢? 分析 我们刚提到多个配置的玄机就在FeignClientProper...

Mr_Qi
11分钟前
1
0
Linux Kernel 4.16 系列停止维护,用户应升级至 4.17

知名 Linux 内核维护人员兼开发人员 Greg Kroah-Hartman 近日在发布 4.16.18 版本的同时,宣布这是 4.16 系列的最后一个维护版本,强烈建议用户立即升级至 4.17 系列。 Linux 4.16 于 2018 年...

问题终结者
35分钟前
0
0
Apache配置时.htaccess失效不起作用的原因分析

.htaccess 失效的原因 1. 重写规则有问题,检查自己的重写规则 2.Apache配置问题,配置中没有配置启用 rewrite a2enmod rewrite 3.网站配置文件没有启用配置需要配置 000-default.conf <Dire...

TU-DESGIN
55分钟前
1
0
两个求最大公约数C/C++算法实现

#include<stdio.h> #include<time.h> #include <iostream>using namespace std;//求最大公约数 LCD(Largest Common Division)//短除法 //m=8251, n=6105; int LCD_ShortDiv(int m, ......

失落的艺术
今天
1
0
QueryPerformanceCounter

windows的Sleep函数,睡眠线程指定毫秒数,可以用来做毫秒延时。 对于微秒延时,没有一个现成的函数,但是可以通过 QueryPerformanceFrequency QueryPerformanceCounter 来间接实现。原理就是...

开飞色
今天
1
0
log4j2使用AsyncRoot不显示行号问题处理

<AsyncRoot level="info" includeLocation="true"> <AppenderRef ref="File"/></AsyncRoot><!--1.异步logger,还需要在pom.xml中添加disruptor的依赖。2.includeLocation结合异......

小翔
今天
3
0
安卓手机上 K 歌,声音延迟怎么解决?

这篇文章可以为你提供一个解决录音和播放同步问题的思路,而且解决了声音从手机传输到耳机上有延时的问题。 初识音频 在开始之前,我先简单介绍一下音频相关的基础知识,方便下文理解。 我们...

编辑部的故事
今天
2
0
使用token实现在有效期内APP自动登录功能

使用token实现在有效期内APP自动登录功能 http://sevennight.cc/2016/07/19/auto_login_impl.html

风云海滩
今天
3
0
Spring Boot集成RabbitMQ发送接收JSON

默认情况下RabbitMQ发送的消息是转换为字节码,这里介绍一下如何发送JSON数据。 ObjectMapper 最简单发送JSON数据的方式是把对象使用ObjectMapper等JSON工具类把对象转换为JSON格式,然后发送...

小致dad
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部