高吞吐量（3GB / s）文件系统可用时，如何使用Java中的多个线程读取文件

Question

我了解对于普通的主轴驱动器系统，使用多个线程读取文件效率很低。

这是另一种情况，我有一个高吞吐量的文件系统可供使用，它具有196个CPU内核和2TB RAM的读取速度高达3GB / s。

单线程Java程序以最大85-100 MB /s的速度读取文件，因此我有可能变得比单线程更好。我必须读取最大1TB的文件，并且有足够的RAM来加载它。

当前，我使用以下内容或类似内容，但需要使用多线程编写内容以获得更好的吞吐量：

Java 7文件：50 MB / s

List<String> lines = Files.readAllLines(Paths.get(path), encoding);

Java Commons-io：48 MB /秒

List<String> lines = FileUtils.readLines(new File("/path/to/file.txt"), "utf-8");

与番石榴相同：45 MB / s

List<String> lines = Files.readLines(new File("/path/to/file.txt"), Charset.forName("utf-8"));

Java扫描程序类：非常慢

Scanner s = new Scanner(new File("filepath"));
ArrayList<String> list = new ArrayList<String>();
while (s.hasNext()){
    list.add(s.next());
}
s.close();

我希望能够以正确的排序顺序尽快加载文件并构建相同的ArrayList。

Answer 1

这是读取具有多个线程的单个文件的解决方案。

_将文件分为N个块，读取线程中的每个块，然后按顺序合并它们。当心跨越块边界的线。这是用户[偷偷]建议的基本思想

下面是对单个20 GB文件的多线程实施的基准测试：

1个线程：50秒：400 MB / s

2个线程：30秒：666 MB / s

4个线程：20秒：1GB / s

8个线程：60秒：333 MB / s

等效的Java7 readAllLines（）：400秒：50 MB / s

注意：这仅适用于旨在支持高吞吐量I / O的系统，不适用于普通的个人计算机

package filereadtests;

import java.io.*;
import static java.lang.Math.toIntExact;
import java.nio.*;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FileRead implements Runnable
{

private FileChannel _channel;
private long _startLocation;
private int _size;
int _sequence_number;

public FileRead(long loc, int size, FileChannel chnl, int sequence)
{
    _startLocation = loc;
    _size = size;
    _channel = chnl;
    _sequence_number = sequence;
}

@Override
public void run()
{
    try
    {
        System.out.println("Reading the channel: " + _startLocation + ":" + _size);

        //allocate memory
        ByteBuffer buff = ByteBuffer.allocate(_size);

        //Read file chunk to RAM
        _channel.read(buff, _startLocation);

        //chunk to String
        String string_chunk = new String(buff.array(), Charset.forName("UTF-8"));

        System.out.println("Done Reading the channel: " + _startLocation + ":" + _size);

    } catch (Exception e)
    {
        e.printStackTrace();
    }
}

//args[0] is path to read file
//args[1] is the size of thread pool; Need to try different values to fing sweet spot
public static void main(String[] args) throws Exception
{
    FileInputStream fileInputStream = new FileInputStream(args[0]);
    FileChannel channel = fileInputStream.getChannel();
    long remaining_size = channel.size(); //get the total number of bytes in the file
    long chunk_size = remaining_size / Integer.parseInt(args[1]); //file_size/threads

    //Max allocation size allowed is ~2GB
    if (chunk_size > (Integer.MAX_VALUE - 5))
    {
        chunk_size = (Integer.MAX_VALUE - 5);
    }

    //thread pool
    ExecutorService executor = Executors.newFixedThreadPool(Integer.parseInt(args[1]));

    long start_loc = 0;//file pointer
    int i = 0; //loop counter
    while (remaining_size >= chunk_size)
    {
        //launches a new thread
        executor.execute(new FileRead(start_loc, toIntExact(chunk_size), channel, i));
        remaining_size = remaining_size - chunk_size;
        start_loc = start_loc + chunk_size;
        i++;
    }

    //load the last remaining piece
    executor.execute(new FileRead(start_loc, toIntExact(remaining_size), channel, i));

    //Tear Down
    executor.shutdown();

    //Wait for all threads to finish
    while (!executor.isTerminated())
    {
        //wait for infinity time
    }
    System.out.println("Finished all threads");
    fileInputStream.close();
}

}