java多线程 之 BlockingQueue

*

本例子来自于Java核心技术卷一，14章多线程

功能：指定一个目录，查找 包含 keyword 的内容

有三个类：

BlockingQueueTest：测试类

FileEnumerationTask：将目录下的文件递归遍历到 BlockingQueue<File> 中

SearchTask：从BlockingQueue<File> 逐个取出来，用scanner 读取

代码：

1，BlockingQueueTest.java

package com.maple.blockingqueue;

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

public class BlockingQueueTest {
    public static void main(String[] args) {
        String directory="e://testfile//";
        String keyword="aaa";
        
        final int FILE_QUEUE_SIZE=5;
        final int SEARCH_THREADS=10;
        BlockingQueue<File> queue=new ArrayBlockingQueue<>(FILE_QUEUE_SIZE);
        
        FileEnumerationTask enumerator=new FileEnumerationTask(queue,new File(directory));
        new Thread(enumerator).start();//将要查询的文件放到队列中
        for(int i=1;i<=SEARCH_THREADS;i++){
            System.out.println("start thread"+i);
            new Thread(new SearchTask(queue,keyword)).start();//从队列中取出文件来扫描
            
        }
    }
}

2,

package com.maple.blockingqueue;

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

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){
            
        }
        
    }
    public void enumerate(File directory) throws InterruptedException{
        File[] files=directory.listFiles();
        for(File file:files){
            if(file.isDirectory())
                enumerate(file);
            else
                queue.put(file);
        }
        
    }

}

3，

package com.maple.blockingqueue;

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

public class SearchTask implements Runnable{

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


    @Override
    public void run() {
        try{
            boolean done=false;
            while(!done){
                File file=queue.take();
                if(file==FileEnumerationTask.DUMMY){
                    queue.put(file);
                    done=true;
                }else
                    search(file);
            }
        }catch(IOException | InterruptedException e){
            e.printStackTrace();
        }
    }
    
    public void search(File file) throws IOException{
        try(Scanner in=new Scanner(file)){
            int lineNumber=0;
            System.out.println("search file:"+file.getName());
            while(in.hasNextLine()){
                lineNumber++;
                String line=in.nextLine();
                System.out.println(keyword+"$$$"+line);
                if(line.contains(keyword))
                    System.out.printf("%s:%d:%s%n",file.getPath(),lineNumber,line);
            }
        }
    }

}

*