多线程-----阻塞队列

使用一个线程将一个指定目录下面的所有文件放在一个阻塞队列中，用一定数量的线程从该队列

中读取文件，然后在该文件中查找指定关键字。该程序展示了阻塞队列的功能。阻塞队列的大小

指定，当队列已满或者为空时，相应向队列放入文件的线程或者从队列中读取文件的线程将会阻

塞。该应用是一个典型的生产者-消费者模式。

源代码如下：

package com.thread.blockQueue;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * this is a BlockingQueue test
 * @author yangjianzhou
 * @description TODO
 * @time Oct 29, 2014 : 9:55:19 PM
 */
public class BlockingQueueTest {

	public static void main(String[] args) {

		String baseDirectory = "/home/yangjianzhou/document";
		String keyword = "java";
		final int FILE_QUEUE_SIZE = 10;
		final int SEARCH_THREADS = 100;

		BlockingQueue<File> queue = new ArrayBlockingQueue<File>(FILE_QUEUE_SIZE);
		FileEnumerationTask enumerator = new FileEnumerationTask(queue, new File(baseDirectory));
		new Thread(enumerator).start();
		for (int i = 1; i <= SEARCH_THREADS; i++) {
			new Thread(new SearchTask(queue, keyword)).start();
		}
	}
}

/**
 * this class is to put  file into the queue
 * @author yangjianzhou
 * @description TODO
 * @time Nov 5, 2014 : 10:30:20 PM
 */
class FileEnumerationTask implements Runnable {

	public static File DUMMY = new File("");

	private BlockingQueue<File> queue;

	private File startingDirectory;

	public FileEnumerationTask(BlockingQueue<File> queue, File startingDirectory) {
		this.queue = queue;
		this.startingDirectory = startingDirectory;
	}

	@Override
	public void run() {
		try {
			enumerate(startingDirectory);
			queue.put(DUMMY);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	/**
	 * enumerate the directory and put the  file into queue and if element is directory 
	 * and enumerate the directory and so on
	 * @param directory
	 * @throws InterruptedException
	 */
	public void enumerate(File directory) throws InterruptedException {
		File[] files = directory.listFiles();
		for (File file : files) {
			if (file.isDirectory()) {
				enumerate(file);
			} else {
				queue.put(file);
			}
		}
	}
}

/**
 * search  keyword in the blockingQueue  that  contains file
 * @author yangjianzhou
 * @description TODO
 * @time Nov 5, 2014 : 10:27:20 PM
 */
class SearchTask implements Runnable {

	private BlockingQueue<File> queue;

	private String keyword;

	public SearchTask(BlockingQueue<File> queue, String keyword) {
		this.queue = queue;
		this.keyword = keyword;
	}

	/**
	 * run the searchTask  thread , this thread gets the blockingQueue and take file from the queue 
	 * and search keyword in the file
	 */
	@Override
	public void run() {

		try {
			boolean done = false;
			while (!done) {
				/*if the queue is empty , this thread is blocked , while the queue is not empty, this thread is to run*/
				File file = queue.take();
				System.out.println("Thread name : " +Thread.currentThread().getName()+" , queue size : "+ queue.size());
				if (file == FileEnumerationTask.DUMMY) {
					queue.put(file);
					done = true;
				} else {
					search(file);
				}
			}
		} catch (Exception e) {

			e.printStackTrace();
		}
	}

	/**
	 * search keyword in the specify file , if the file contains the given keyword , print the line info
	 * @param file
	 * @throws IOException
	 */
	public void search(File file) throws IOException {
		Scanner in = new Scanner(new FileInputStream(file));
		int lineNumber = 0;
		while (in.hasNextLine()) {
			lineNumber++;
			String line = in.nextLine();
			if (line.contains(keyword)) {
				System.out.printf("%s;%d;%s%n", file.getPath(), lineNumber, line);
			}
		}
		in.close();
	}

}

运行结果：

Thread name : Thread-2 , queue size : 4
Thread name : Thread-26 , queue size : 2
Thread name : Thread-27 , queue size : 2
Thread name : Thread-25 , queue size : 4
Thread name : Thread-1 , queue size : 1
Thread name : Thread-53 , queue size : 0
Thread name : Thread-24 , queue size : 0
Thread name : Thread-56 , queue size : 9
Thread name : Thread-30 , queue size : 10
Thread name : Thread-57 , queue size : 7
Thread name : Thread-28 , queue size : 7
Thread name : Thread-58 , queue size : 6
Thread name : Thread-10 , queue size : 5
Thread name : Thread-59 , queue size : 3
Thread name : Thread-60 , queue size : 2
Thread name : Thread-61 , queue size : 1
Thread name : Thread-55 , queue size : 0
Thread name : Thread-91 , queue size : 0
Thread name : Thread-62 , queue size : 0
Thread name : Thread-96 , queue size : 9
Thread name : Thread-10 , queue size : 8
Thread name : Thread-50 , queue size : 8
Thread name : Thread-51 , queue size : 8
Thread name : Thread-6 , queue size : 8
Thread name : Thread-52 , queue size : 8
Thread name : Thread-49 , queue size : 8
Thread name : Thread-48 , queue size : 5
Thread name : Thread-47 , queue size : 6
Thread name : Thread-46 , queue size : 7
Thread name : Thread-45 , queue size : 8
Thread name : Thread-44 , queue size : 9
Thread name : Thread-7 , queue size : 10
Thread name : Thread-4 , queue size : 7
Thread name : Thread-2 , queue size : 8
Thread name : Thread-100 , queue size : 6
Thread name : Thread-35 , queue size : 0
Thread name : Thread-54 , queue size : 7
Thread name : Thread-42 , queue size : 0
Thread name : Thread-8 , queue size : 6
Thread name : Thread-41 , queue size : 1
Thread name : Thread-63 , queue size : 5