您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 多線程大文件分割：RandomAccessFile + CountDownLatch 秒級處理 GB 級 CSV

編輯：Java 相關知識

面對數 GB 的 CSV 文件導入數據庫，單線程讀寫極慢且易 OOM。本文實現一個多線程分割工具，採用預分配文件空間 + RandomAccessFile 並發讀寫 + CountDownLatch 同步 + 行級合併修復，將 100 MB 文件分割耗時壓縮至毫秒級，並完美處理跨分片行截斷問題。

一、核心設計

分割策略：按固定字節大小（如 64 MB，2 的 N 次方避免中文截斷）劃分分片
並發讀寫：每個線程獨立 RandomAccessFile 讀取指定偏移量，寫入獨立分片文件
同步等待：CountDownLatch 確保所有分片寫入完成
行修復：分割可能截斷行，合併階段檢測首尾行分隔符數量，補全跨片行

二、完整實現代碼

public class CsvSplitter {
    private static final long SPLIT_SIZE = 64 * 1024 * 1024; // 64 MB
    private static final char CSV_SEP = '^';  // CSV 分隔符
    private static final Logger LOG = LoggerFactory.getLogger(CsvSplitter.class);
    
    public List split(File source) throws IOException, InterruptedException {
        if (source.length() < SPLIT_SIZE) return Collections.singletonList(source.getAbsolutePath());
        
        int parts = (int) Math.ceil((double) source.length() / SPLIT_SIZE);
        CountDownLatch latch = new CountDownLatch(parts);
        List partFiles = new ArrayList<>(parts);
        
        // 1. 並發分割
        for (int i = 0; i < parts; i++) {
            long start = i * SPLIT_SIZE;
            long end = Math.min((i + 1) * SPLIT_SIZE, source.length());
            int readSize = (int) (end - start);
            String partName = source.getAbsolutePath() + "." + String.format("%03d", i+1) + ".csv";
            partFiles.add(partName);
            
            new Thread(new SplitTask(source, start, readSize, partName, latch)).start();
        }
        
        latch.await(); // 等待所有分片完成
        
        // 2. 合併修復跨分片行
        mergeRows(partFiles);
        return partFiles;
    }
    
    private static class SplitTask implements Runnable {
        private final File source;
        private final long startPos;
        private final int readSize;
        private final String partFile;
        private final CountDownLatch latch;
        
        SplitTask(File source, long startPos, int readSize, String partFile, CountDownLatch latch) {
            this.source = source; this.startPos = startPos; this.readSize = readSize;
            this.partFile = partFile; this.latch = latch;
        }
        
        @Override public void run() {
            try (RandomAccessFile raf = new RandomAccessFile(source, "r");
                 FileOutputStream fos = new FileOutputStream(partFile)) {
                
                raf.seek(startPos);
                byte[] buf = new byte[readSize];
                int len = raf.read(buf);
                fos.write(buf, 0, len);
            } catch (IOException e) {
                LOG.error("分片寫入失敗: {}", partFile, e);
            } finally {
                latch.countDown();
            }
        }
    }
    
    // 行修復邏輯
    private void mergeRows(List partFiles) throws IOException {
        List infos = new ArrayList<>();
        
        // 收集每個分片首行、尾行、分隔符計數
        for (int i = 0; i < partFiles.size(); i++) {
            PartInfo info = new PartInfo(partFiles.get(i));
            try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(info.file), "GBK"))) {
                info.firstRow = br.readLine();
                String line;
                while ((line = br.readLine()) != null) info.lastRow = line;
            }
            if (i > 0) {
                String prevLast = infos.get(i-1).lastRow;
                // 判斷首行是否為上一片尾行的續行：分隔符數量對比
                info.firstRowComplete = countSep(info.firstRow) > countSep(prevLast + info.firstRow);
            }
            infos.add(info);
        }
        
        // 寫入修復內容
        for (int i = 0; i < infos.size() - 1; i++) {
            PartInfo cur = infos.get(i);
            PartInfo next = infos.get(i+1);
            StringBuilder fix = new StringBuilder();
            if (next.firstRowComplete) fix.append("\r\n");
            fix.append(next.firstRow);
            appendToFile(cur.file, fix.toString());
        }
    }
    
    private int countSep(String s) { /* 統計 CSV_SEP 出現次數 */ }
    private void appendToFile(String file, String content) { /* RandomAccessFile 定位末尾追加 */ }
    
    static class PartInfo {
        File file; String firstRow, lastRow; boolean firstRowComplete;
        PartInfo(String path) { this.file = new File(path); }
    }
}