BlockingQueue实现生产者消费者模式
?package com.sunrise.mywork2.concurrent.queue;
import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class IndexingService
{
private static final int CAPACITY = 1000;
private static final File POISON = new File("");
private final IndexerThread consumer = new IndexerThread();
private final CrawlerThread producer = new CrawlerThread();
private final BlockingQueue<File> queue;
private final FileFilter fileFilter;
private final File root;
public IndexingService(File root,final FileFilter fileFilter){
?this.root = root;
?this.queue = new LinkedBlockingQueue<File>(CAPACITY);
?this.fileFilter = new FileFilter()
?{
? public boolean accept(File pathname)
? {
? ?return pathname.isDirectory()||fileFilter.accept(pathname);
? }
?};
}
private boolean alreadyIndexed(File f){
?return false;
}
class CrawlerThread extends Thread{
?public void run(){
? try{
? ?crawl(root);
? }catch (InterruptedException e) {
? }finally{
? ?while(true){
? ? try{
? ? ?queue.put(POISON);
? ? ?break;
? ? }catch (InterruptedException e1) {
? ? }
? ?}
? }
?}
?private void crawl(File root) throws InterruptedException{
? File[] entries = root.listFiles(fileFilter);
? if(entries!=null){
? ?for(File entry:entries){
? ? if(entry.isDirectory())
? ? ?crawl(entry);
? ? else if(!alreadyIndexed(entry))
? ? ?queue.put(entry);
? ?}
? }
?}
}
class IndexerThread extends Thread{
?public void run(){
? try{
? ?while(true){
? ? File file = queue.take();
? ? if(file == POISON)
? ? ?break;
? ? else
? ? ?indexFile(file);
? ?}
? }catch (InterruptedException e) {
? }
?}
?public void indexFile(File file){
? //TODO
? System.out.println(file.getName());
?}
}
public void start(){
?producer.start();
?consumer.start();
}
public void stop(){
?producer.interrupt();
}
public void awaitTermination() throws InterruptedException{
?consumer.join();
}
/**
?* test it
?* @param args
?*/
public static void main(String[] args)
{
?File root = new File("F:\");
?FileFilter fileFilter = new FileFilter()
?{
? public boolean accept(File f)
? {
? ?return f.getName().lastIndexOf(".java")>-1;
? }
?};
?IndexingService is = new IndexingService(root, fileFilter);
?is.start();
}
}
