文档章节

Java并发(四)BlockingQueue的使用

摆渡者
 摆渡者
发布于 2015/10/12 10:57
字数 1222
阅读 592
收藏 5

    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通信。

© 著作权归作者所有

共有 人打赏支持
摆渡者
粉丝 326
博文 171
码字总数 205876
作品 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面试:投行的15个多线程和并发面试题

本文由ImportNew -一杯哈希不加盐 翻译自dzone。欢迎加入翻译小组。转载请见文末要求。 多线程和并发问题已成为各种 Java 面试中必不可少的一部分。如果你准备参加投行的 Java 开发岗位面试,...

ImportNew
08/23
0
0
【死磕Java并发】—– 死磕 Java 并发精品合集

【死磕 Java 并发】系列是 LZ 在 2017 年写的第一个死磕系列,一直没有做一个合集,这篇博客则是将整个系列做一个概览。 先来一个总览图: 【高清图,请关注“Java技术驿站”公众号,回复:脑...

chenssy
07/22
0
0

没有更多内容

加载失败,请刷新页面

加载更多

如何通过 J2Cache 实现分布式 session 存储

做 Java Web 开发的人多数都会需要使用到 session (会话),我们使用 session 来保存一些需要在两个不同的请求之间共享数据。一般 Java 的 Web 容器像 Tomcat、Resin、Jetty 等等,它们会在...

红薯
今天
1
0
C++ std::thread

C++11提供了std::thread类来表示一个多线程对象。 1,首先介绍一下std::this_thread命名空间: (1)std::this_thread::get_id():返回当前线程id (2)std::this_thread::yield():用户接口...

yepanl
今天
2
0
Nignx缓存文件与动态文件自动均衡的配置

下面这段nginx的配置脚本的作用是,自动判断是否存在缓存文件,如果有优先输出缓存文件,不经过php,如果没有,则回到php去处理,同时生成缓存文件。 PHP框架是ThinkPHP,最后一个rewrite有关...

swingcoder
今天
1
0
20180920 usermod命令与用户密码管理

命令 usermod usermod 命令的选项和 useradd 差不多。 一个用户可以属于多个组,但是gid只有一个;除了gid,其他的组(groups)叫做扩展组。 usermod -u 1010 username # 更改用户idusermod ...

野雪球
今天
1
0
Java网络编程基础

1. 简单了解网络通信协议TCP/IP网络模型相关名词 应用层(HTTP,FTP,DNS等) 传输层(TCP,UDP) 网络层(IP,ICMP等) 链路层(驱动程序,接口等) 链路层:用于定义物理传输通道,通常是对...

江左煤郎
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部