Java利用多線程和分塊實(shí)現(xiàn)快速讀取文件

更新時(shí)間：2023年09月08日 09:12:18 作者：L7ink

在工作中經(jīng)常會(huì)有接收文件并且讀取落庫(kù)的需求,讀取方式都是串行讀取,所以本文主要為大家介紹一下如何利用多線程和分塊實(shí)現(xiàn)快速讀取文件,希望對(duì)大家有所幫助

背景

在工作中經(jīng)常會(huì)有接收文件并且讀取落庫(kù)的需求，讀取方式都是串行讀取，即一行行的讀取，如果文件小還可以，但是如果文件比較大，類似于全量文件的話，這樣的讀取就會(huì)非常效率低。

本文主要介紹的是如何正確的將文件分塊，多線程的實(shí)現(xiàn)方式有多種，這里用的是CompletableFuture

方法

因?yàn)槲覀兾募锩娴拿織l數(shù)據(jù)之間沒(méi)有任何依賴關(guān)系也不存在順序要求。如何提高讀取速度，第一個(gè)想到當(dāng)然就是并行讀取文件，并行讀取的前提就是要給文件分塊，讓每個(gè)線程只讀取對(duì)應(yīng)分塊的數(shù)據(jù)，先看看我們的文件格式

可以看見(jiàn)我們的文件格式每一行的長(zhǎng)度不一，同時(shí)文件也無(wú)法像TCP通過(guò)指定數(shù)據(jù)體的長(zhǎng)度來(lái)讀取數(shù)據(jù)，所以如何能正確的分塊是整個(gè)方法的關(guān)鍵，如果分多或者分少了就會(huì)導(dǎo)致數(shù)據(jù)讀取錯(cuò)誤的可能。

可以看見(jiàn)錯(cuò)誤的分塊就會(huì)導(dǎo)致我們讀取的數(shù)據(jù)會(huì)被截取掉一部分，截取掉多少都是隨機(jī)的。這里我們用的方法是用填充來(lái)讓每個(gè)分塊都是正確的。具體來(lái)說(shuō)就是我們?cè)诜謮K的時(shí)候，判斷一下當(dāng)前的分塊位置會(huì)不會(huì)導(dǎo)致數(shù)據(jù)被截取，因?yàn)槲覀兊臄?shù)據(jù)是一行行的，所以最好的分塊位置都是分在了行尾。如果說(shuō)當(dāng)前的分塊位置是在一行的中間的話，那我們就要移動(dòng)這個(gè)分塊的位置到這行的行尾去

private static int THREAD_NUM = 5;
long total = file.length();
long chunkSize = total < THREAD_NUM ? total : total / THREAD_NUM;

先確定要用幾個(gè)線程并行讀取，然后根據(jù)線程數(shù)和文件的大小來(lái)確定每一塊的大小，接下來(lái)就進(jìn)行判斷是否需要填充

    private static long padding(long start, long chunkSize, File file) {
        try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")){
            randomAccessFile.seek(start + chunkSize);
            boolean eol = false;
            //判斷當(dāng)前的位置是否需要填充,如果當(dāng)前沒(méi)有數(shù)據(jù)或者是行尾,則不需要填充
            switch (randomAccessFile.read()) {
                case -1:
                case '\n':
                    eol = true;
                    break;
                case '\r':
                    eol = true;
                    break;
                default:
                    break;
            }
            //如果符合填充條件，對(duì)其進(jìn)行填充，首先是讀取一行數(shù)據(jù)，然后計(jì)算出這行數(shù)據(jù)的長(zhǎng)度，然后將這行數(shù)據(jù)的長(zhǎng)度加上前面讀取了一字節(jié)，然后將這些長(zhǎng)度加到chunkSize上
            if (!eol){
                String readLine = randomAccessFile.readLine();
                chunkSize += readLine.getBytes().length;
                //加上前面讀取的一字節(jié)
                chunkSize += 1;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return chunkSize;
    }

如何填充的可以看看代碼注釋，是參照了readline的實(shí)現(xiàn)思路。經(jīng)過(guò)填充后就可以確保每塊的分塊的最后一個(gè)位置都是在行尾。將每個(gè)分塊的起始位置和分塊大小儲(chǔ)存起來(lái)再結(jié)合上CompletableFuture就可以多線程分塊讀取文件了

        Map<Long, Long> chunkMap = new HashMap<>();
        for (int i = 0; i < THREAD_NUM; i++) {
            chunkSize = padding(start, chunkSize, file);
            chunkMap.put(start, chunkSize);
            start += chunkSize;
        }
        CompletableFuture.allOf(chunkMap.entrySet().stream().map(entry -> CompletableFuture.runAsync( () -> handlerReportTreeBaseData(entry.getKey(), entry.getValue()))).toArray(CompletableFuture[]::new))
                .exceptionally(throwable -> {
                    System.out.println(throwable.getMessage());
                    return null;
                }).join();

完整實(shí)現(xiàn)

接下來(lái)給出整個(gè)的實(shí)現(xiàn)代碼，歡迎大家看看有沒(méi)有什么我沒(méi)有考慮到的，有可能的隱藏BUG和還能優(yōu)化改善的地方，歡迎討論

public class SpiltFIle {
    private static int THREAD_NUM = 5;
    private static void splitChunks() {
        File file = new File("test.txt");
        long total = file.length();
        long chunkSize = total < THREAD_NUM ? total : total / THREAD_NUM;
        long start = 0;
        Map<Long, Long> chunkMap = new HashMap<>();
        for (int i = 0; i < THREAD_NUM; i++) {
            chunkSize = padding(start, chunkSize, file);
            handlerReportTreeBaseData(start, chunkSize);
            chunkMap.put(start, chunkSize);
            start += chunkSize;
        }
        CompletableFuture.allOf(chunkMap.entrySet().stream().map(entry -> CompletableFuture.runAsync( () -> handlerReportTreeBaseData(entry.getKey(), entry.getValue()))).toArray(CompletableFuture[]::new))
                .exceptionally(throwable -> {
                    System.out.println(throwable.getMessage());
                    return null;
                }).join();
    }
    private static long padding(long start, long chunkSize, File file) {
        try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")){
            randomAccessFile.seek(start + chunkSize);
            boolean eol = false;
            //判斷當(dāng)前的位置是否需要填充,如果當(dāng)前沒(méi)有數(shù)據(jù)或者是行尾,則不需要填充
            switch (randomAccessFile.read()) {
                case -1:
                case '\n':
                    eol = true;
                    break;
                case '\r':
                    eol = true;
                    break;
                default:
                    break;
            }
            //如果符合填充條件，對(duì)其進(jìn)行填充，首先是讀取一行數(shù)據(jù)，然后計(jì)算出這行數(shù)據(jù)的長(zhǎng)度，然后將這行數(shù)據(jù)的長(zhǎng)度加上前面讀取了一字節(jié)，然后將這些長(zhǎng)度加到chunkSize上
            if (!eol){
                String readLine = randomAccessFile.readLine();
                chunkSize += readLine.getBytes().length;
                chunkSize += 1; //加上前面讀取的一字節(jié)
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return chunkSize;
    }
    private static void handlerReportTreeBaseData(long start, long chunkSize) {
        try (RandomAccessFile randomAccessFile = new RandomAccessFile("test.txt", "r")) {
            randomAccessFile.seek(start);
            long currentCount = 0L;
            String line;
            while (currentCount < chunkSize && (line = randomAccessFile.readLine()) != null){
                if (!line.isEmpty()){
                    currentCount += line.getBytes().length + System.lineSeparator().getBytes().length;
                    System.out.println(line);
                }
            }
        }catch (Exception ignored){
        }
    }
    public static void main(String[] args) throws IOException {
        SpiltFIle.splitChunks();
    }
}

最后也是能正常的讀取完文件