并发编程 03—— 阻塞队列和生产者-消费者模式

Java并发编程实践 目录

并发编程 01—— ThreadLocal

并发编程 02—— ConcurrentHashMap

并发编程 03—— 阻塞队列和生产者-消费者模式

并发编程 04—— 闭锁CountDownLatch 与 栅栏CyclicBarrier

并发编程 05—— Callable和Future

并发编程 06—— CompletionService : Executor 和 BlockingQueue

并发编程 07—— 任务取消

并发编程 08—— 任务取消 之 中断

并发编程 09—— 任务取消 之 停止基于线程的服务

并发编程 10—— 任务取消 之 关闭 ExecutorService

并发编程 11—— 任务取消 之 “毒丸”对象

并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性

并发编程 13—— 线程池的使用 之 配置ThreadPoolExecutor 和 饱和策略

并发编程 14—— 线程池 之 整体架构

并发编程 15—— 线程池 之 原理一

并发编程 16—— 线程池 之 原理二

并发编程 17—— Lock

并发编程 18—— 使用内置条件队列实现简单的有界缓存

并发编程 19—— 显式的Conditon 对象

并发编程 20—— AbstractQueuedSynchronizer 深入分析

并发编程 21—— 原子变量和非阻塞同步机制

概述

第1部分 为什么要使用生产者和消费者模式

第2部分 什么是生产者消费者模式

第3部分 代码示例

第1部分 为什么要使用生产者和消费者模式

　　在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

第2部分 什么是生产者消费者模式

　　生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式，都会找一个第三者出来进行解耦，如工厂模式的第三者是工厂类，模板模式的第三者是模板类。在学习一些设计模式的过程中，如果先找到这个模式的第三者，能帮助我们快速熟悉一个设计模式。

　　虽然生产者-消费者模式能够将生产者和消费者的代码彼此解耦开来，但它们的行为仍然会通过共享工作队列间接地耦合在一起。开发人员总是会假设消费者处理工作的速率能赶上生产者生成工作项的速率，因此通常不会为工作队列的大小设置边界，但这将导致在之后需要重新设计系统架构。因此，应该尽早地通过阻塞队列在设计中构建资源管理机制——这件事做得越早，就越容易。

　　在类库中包含了BlockingQueue的多种实现，其中，LinkedBlockingQueue和ArrayBlockingQueue是FIFO队列，二者分别与LinkedList和ArrayList类似，但比同步List拥有更好的并发性能。PriorityBlockingQueue是一个按优先级排序的队列，当不希望按照某种顺序而不是FIFO来处理元素时，这个队列将非常有用。

第3部分 代码示例

　　有一种类型的程序适合被分解为生产者和消费者，例如代理程序，它将扫描本地驱动上的文件并建立索引以便随后进行搜索，类似于某些桌面搜索程序或者Windows索引服务。在下面程序中，FileCrawler 给出一个生产者任务，即在某个文件层次结构中搜索符合索引标准的文件，并将它们的名称放入工作队列。而且，在Indexer中给出一个消费者任务，即从队列中取出文件的名称并对它们建立索引。

package com.concurrency.BasicBuildingBlocks_5;

import java.io.File;
import java.util.concurrent.BlockingQueue;

/**
* 5.8 Producer and consumer tasks in a desktop search application
*
* @ClassName: ProducerConsumer
* @author Xingle
* @date 2014-9-3 下午3:43:49
*/
public class ProducerConsumer {

//生产者任务，在某个文件层次结构中搜索符合索引标准的文件，并将它们放入工作队列
static class FileCrawler implements Runnable {
private final BlockingQueue<File> fileQueue;
private final File root;

public FileCrawler(BlockingQueue<File> fileQueue,File root) {
this.fileQueue = fileQueue;

this.root = root;

}

@Override
public void run() {
crawl(root);
}

/**
* 搜索没有被索引过的文件放入阻塞队列
* @author xingle
* @data 2014-9-3 下午5:13:52
*/
private void crawl(File root) {
File[] entries = root.listFiles();
if (entries != null) {
for (File entrie : entries) {
if (entrie.isDirectory()) {
crawl(entrie);
} else {
if (!alreadyIndexed(entrie)){
System.out.println("放入生产者队列文件："+entrie.getName()+"来自线程："+Thread.currentThread().getName());
fileQueue.add(entrie);
}

}
}
}
}

/**
* 是否已经被索引
* @param entrie
* @return
* @author xingle
* @data 2014-9-3 下午5:26:04
*/
private boolean alreadyIndexed(File entrie) {
return false;
}

}

//消费者
static class Indexer implements Runnable {
private final BlockingQueue<File> queue;

public Indexer(BlockingQueue<File> queue){
this.queue = queue;
}
@Override
public void run() {
while(true){
try {
indexFile(queue.take());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

private void indexFile(File file) {
System.out.println("消费者取出文件："+file.getName()+"来自线程："+Thread.currentThread().getName());
};
}

}

测试程序：

package com.concurrency.BasicBuildingBlocks_5;

import java.io.File;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.concurrency.BasicBuildingBlocks_5.ProducerConsumer.FileCrawler;
import com.concurrency.BasicBuildingBlocks_5.ProducerConsumer.Indexer;

/**
* 测试生成者和消费者
* @ClassName: testMain
* TODO
* @author Xingle
* @date 2014-9-3 下午6:03:56
*/
public class testMain {

public static void main(String[] args) {

File file = new File("D:\\test1/");
File[] roots = file.listFiles();
final int BOUND = 10;
final int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
BlockingQueue<File> queue = new LinkedBlockingQueue<File>(BOUND);

//消费
for (int i = 0; i < N_CONSUMERS; i++) {
new Thread(new Indexer(queue)).start();
}
//生产
for (File root : roots) {
new Thread(new FileCrawler(queue, root)).start();
}
}

}

执行结果：



参考：

1.《Java 并发编程实战》5.3节

2.聊聊并发——生产者消费者模式