java - 如何在多进程、多线程环境下读取文件

Question

我在使用文件锁定的多进程环境中成功读取文件在多线程(单进程)的情况下，我使用了一个用文件名填充的队列，单独打开一个线程，从中读取然后等到整个读取结束，之后我习惯重命名它们。通过这种方式，我过去常常以多线程方式(批量)读取文件。

现在，我想使用多进程和多线程读取目录中的文件。我尝试合并我的两种方法，但效果不佳。日志显示很多文件显示 FileNotFound 异常(因为它们的名称被更改)，一些从未被读取(因为线程死了)，有时锁没有被释放。

 ///////////////////////////////////////////////////////////////////////
     //file filter inner class
     class myfilter implements FileFilter{

            @Override
            public boolean accept(File pathname) {
                // TODO Auto-generated method stub
                Pattern pat = Pattern.compile("email[0-9]+$");
                Matcher mat = pat.matcher(pathname.toString());
                if(mat.find()) {
                    return true;
                }
                return false;
            }

        }
     /////////////////////////////////////////////////////////////////////////


    myfilter filter = new myfilter();
    File alreadyread[] = new File[5];
    Thread t[] = new Thread[5];
    fileread filer[] = new fileread[5];
    File file[] = directory.listFiles(filter);
    FileChannel filechannel[] = new FileChannel[5];
    FileLock lock[] = new FileLock[5];
    tuple_json = new ArrayList();
    //System.out.println("ayush");
    while(true) {
        //declare a queue
        ConcurrentLinkedQueue filequeue = new ConcurrentLinkedQueue();

        //addfilenames to queue and their renamed file names
        try{
        if(file.length!=0) {
            //System.out.println(file.length);
            for(int i=0;i<5 && i<file.length;i++) {

                System.out.println("acquiring lock on file " + file[i].toString());
                try{
                filechannel[i] = new RandomAccessFile(file[i], "rw").getChannel();
                lock[i] = filechannel[i].tryLock();
                }
                catch(Exception e) {
                    file[i] = null;
                    lock[i] = null;
                    System.out.println("cannot acquire lock");
                }
                if(lock[i]!=null){
                    System.out.println("lock acquired on file " + file[i].toString());
                   filequeue.add(file[i]);
                   alreadyread[i] = new File(file[i].toString() + "read");
                   System.out.println(file[i].toString() + "-----" + times);
                }
                else{

                   System.out.println("else condition of acquiring lock");
                    file[i] = null;
                }
                System.out.println("-----------------------------------");
            }

        //starting the thread to read the files
        for(int i=0;i<5 && i<file.length && lock[i]!=null && file[i]!=null;i++){
            filer[i] = new fileread(filequeue.toArray()[i].toString());
            t[i] = new Thread(filer[i]);
            System.out.println("starting a thread to read file" + file[i].toString());
            t[i].start();
        }

        //read the text
        for(int i=0;i<5 && i<file.length && lock[i]!=null && file[i]!=null;i++) {
            try {
                System.out.println("waiting to read " + file[i].toString() + " to be read completely");
                t[i].join();
                System.out.println(file[i] + " was read completetly");
                //System.out.println(filer[i].getText());

            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }

        }
        //file has been read Now rename the file
        for(int i=0;i<5 && i<file.length && lock[i]!=null && file[i]!=null;i++){
            if(lock[i]!=null){
                System.out.println("renaming file " + file[i].toString());
                file[i].renameTo(alreadyread[i]);
                System.out.println("releasing lock on file  " + file[i].toString());
                lock[i].release();
        }
        }

//rest of the processing
/////////////////////////////////////////////////////////////////////////////////////////////////////

文件读取类

class fileread implements Runnable{
//String loc = "/home/ayusun/workspace/Eclipse/fileread/bin";
String fileloc;
BufferedReader br;
String text = "";
public fileread(String filename) {
    this.fileloc = filename;
}
@Override
public void run() {
    try {
        br = new BufferedReader(new FileReader(fileloc));
        System.out.println("started reading file" + fileloc);
        String currline;
        while((( currline = br.readLine())!=null)){
            if(text == "")
                text += currline;
            else
                text += "\n" + currline;
        }

        System.out.println("Read"  + fileloc  + " completely");
        br.close();

    } catch ( IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

}

public String getText() {
    return text;
}
}

我想知道，是否还有其他我可以采用的方法。

Answer 1

如果你想创建对文件的独占访问，你不能使用文件锁定，因为在大多数操作系统上文件锁定是建议性的，而不是强制性的。

我建议为您的所有进程创建一个公共(public)锁目录；在这个锁定目录中，您可以在打开文件之前为每个要锁定的文件创建一个目录。

最大的优势在于目录创建与文件创建不同，它是原子的；因此，您可以使用 Files.createDirectory()(如果您仍然使用 Java6，则可以使用 File 的 .mkdir()，但不要忘记检查返回码)来锁定您阅读的文件。如果失败，则您知道其他人正在使用该文件。

当然，当你处理完一个文件后，不要忘记删除与这个文件匹配的锁目录...(在 finally block 中)

(注意:在 Java 7 中，您可以使用 Files.newBufferedReader()；甚至还有 Files.readAllLines())