多个线程从同一个文件/奇怪的行为中读取答案

【问题标题】：Multiple threads reading from same file/odd behavior多个线程从同一个文件/奇怪的行为中读取
【发布时间】：2014-03-28 03:16:12
【问题描述】：

我正在编写一个程序，该程序需要从一个非常大的文件（400K+ 行）中读取行并将每行中的数据发送到 Web 服务。我决定尝试线程并看到一些我没有预料到的行为，看起来我的 BufferedReader 开始重用它在我调用 readline() 时已经给我的行。

我的程序由两个类组成。一个“主”类，它启动线程并持有对 BufferedReader 的静态引用，并具有一个静态同步的“readNextLine()”方法，线程可以使用该方法基本上调用 BufferedReder 上的 readLine()。以及调用 readNextLine() 并使用来自每个 readNextLine() 调用的数据进行 Web 服务调用的“Runnable”类。我将 BufferedReader 和 readNextLine() 设为静态只是因为这是我能想到的让线程共享阅读器的唯一方法，除了将我的主类的实例传递给线程之外，我不确定哪个更好。

大约 5 分钟后，我开始在我的 Web 服务中看到错误，说它正在处理已经处理过的行。我能够验证线路确实被多次发送，相隔几分钟。

有没有人知道为什么 BufferedReader 似乎给出了它已经读取的线程行？我的印象是 readline() 是连续的，我需要做的就是确保对 readline() 的调用是同步的。

我将在下面展示一些 Main 类代码。 runnable 本质上是一个 while 循环，它调用 readNextLine() 并处理每一行，直到没有剩下的行。

主类：

 //showing reader and thread creation
 inputStream = sftp.get(path to file);
 reader = new BufferedReader(new InputStreamReader(inputStream));

 ExecutorService executor = Executors.newFixedThreadPool(threads);
        Collection<Future> futures = new ArrayList<Future>();
        for(int i=0;i<threads;i++){
        MyRunnable runnable = new   MyRunnable(i);
            futures.add(executor.submit(runnable));
        }
        LOGGER.debug("futures.get()");
        for(Future f:futures){
            f.get();  //use to wait until all threads are done 
        }

public synchronized static String readNextLine(){
    String results = null;
    try{
        if(reader!=null){
         results = reader.readLine();
        }
    }catch(Exception e){
        LOGGER.error("Error reading from file");
    }

    return results;
}

【问题讨论】：

我认为您需要使用 RandomAccessFile 并让每个线程从不同的偏移量读取，尽管我会使用单个线程来读取文件的块，并且对于读取的每个块，拆分多个线程用部分块联系你的网络服务。
其实我只是偶然发现了这一点，docs.oracle.com/javase/7/docs/api/java/nio/channels/… 我认为如果 Java 7 是一个选项，它可能会做你想做的事情。

标签： java multithreading

【解决方案1】：

我正在测试您所说的内容，但我发现您的 readNextLine() 方法中出现错误逻辑，如何调用 reader.readLine()，因为结果为 null 且 if 条件不为 null？

现在我完成了我的demo，看起来效果不错，下面是demo，没有出现re-read line：

 static BufferedReader reader;

public static void main(String[] args) throws FileNotFoundException, ExecutionException, InterruptedException {
    reader = new BufferedReader(new FileReader("test.txt"));
    ExecutorService service = Executors.newFixedThreadPool(3);
    List<Future> results = new ArrayList<Future>();
    for (int i = 0; i < 3; i++) {
        results.add(service.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    String line = null;
                    while ((line = readNextLine()) != null) {
                        System.out.println(line);
                    }
                } catch (IOException e) {
                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                }
            }
        }));
    }
}

public synchronized static String readNextLine() throws IOException {
    return reader.readLine();
}

【讨论】：