java - 通过并发增加磁盘读取吞吐量

Question

我正在尝试读取一个日志文件并解析它，它只消耗 CPU。我有一台服务器以每秒 230MB 的速度读取一个巨大的文本文件，只读取文本文件而不解析。当我尝试使用单线程解析文本文件时，我可以以大约 50-70MB/秒的速度解析文件。

我想增加我的吞吐量，做那个工作并发。在这段代码中，我达到了 130 MB/秒。在峰值时，我看到了 190MB/秒。我尝试了 BlockedQueue、Semaphore、ExecutionService 等。您有什么建议可以让我达到 200MB/秒的吞吐量。

public static void fileReaderTestUsingSemaphore(String[] args) throws Exception {

    CustomFileReader reader = new CustomFileReader(args[0]);
    final int concurrency = Integer.parseInt(args[1]);
    ExecutorService executorService = Executors.newFixedThreadPool(concurrency);
    Semaphore semaphore = new Semaphore(concurrency,true);
    System.out.println("Conccurrency in Semaphore: " + concurrency);


    String line;

    while ((line = reader.getLine()) != null)
    {
        semaphore.acquire();

        try
        {

            final String p = line;

            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    reader.splitNginxLinewithIntern(p); // that is the method which parser string and convert to class.
                    semaphore.release();
                }
            });
        }

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

        finally {
            semaphore.release();
        }
    }

    executorService.shutdown();
    executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);

    System.out.println("ReadByteCount: " + reader.getReadByteCount());
}

Answer 1

您可能会受益于 Files.lines() 方法和 Java 8 中引入的 Stream 范式。它将使用系统通用的 fork/join 池。试试这个模式:

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

public class LineCounter
{
    public static void main(String[] args) throws IOException
    {
        Files.lines(Paths.get("/your/file/here"))
             .parallel()
             .forEach(LineCounter::processLine);
    }

    private static void processLine(String line) {
        // do the processing
    }
}