Java文件导入提速案例实操

Java文件导入提速案例实操

CSV文件导入优化案例

优化前：逐行逐字段处理

public class SlowCSVImport {
    public void importCSV(String filePath) {
        try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
            String line;
            List<DataRecord> records = new ArrayList<>();
            // 问题1：逐行读取，每次都要解析
            while ((line = reader.readLine()) != null) {
                String[] fields = line.split(",");
                DataRecord record = new DataRecord();
                // 问题2：逐字段解析，包含大量字符串操作
                record.setId(Long.parseLong(fields[0].trim()));
                record.setName(fields[1].trim());
                record.setAge(Integer.parseInt(fields[2].trim()));
                // ... 更多字段处理
                // 问题3：逐条写入数据库
                saveToDatabase(record);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private void saveToDatabase(DataRecord record) {
        // 每次创建新的数据库连接
        try (Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "user", "pass")) {
            PreparedStatement ps = conn.prepareStatement("INSERT INTO records VALUES (?, ?, ?)");
            ps.setLong(1, record.getId());
            ps.setString(2, record.getName());
            ps.setInt(3, record.getAge());
            ps.executeUpdate();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

优化后：批量处理 + 多线程

public class OptimizedCSVImport {
    private static final int BATCH_SIZE = 1000;
    private static final int THREAD_COUNT = Runtime.getRuntime().availableProcessors();
    private static final ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
    public void importCSV(String filePath) {
        try {
            long startTime = System.currentTimeMillis();
            // 使用FileChannel进行快速文件读取
            FileChannel channel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ);
            // 创建线程安全的队列
            BlockingQueue<List<DataRecord>> batchQueue = new LinkedBlockingQueue<>(100);
            // 启动写入线程
            List<Future<Integer>> futures = new ArrayList<>();
            for (int i = 0; i < THREAD_COUNT; i++) {
                futures.add(executor.submit(new DatabaseWriter(batchQueue)));
            }
            // 读取文件
            readFileInBatches(channel, batchQueue);
            // 发送完成信号
            for (int i = 0; i < THREAD_COUNT; i++) {
                batchQueue.put(new ArrayList<>(0)); // 空列表表示完成
            }
            // 等待所有写入完成
            int totalRecords = 0;
            for (Future<Integer> future : futures) {
                totalRecords += future.get();
            }
            long endTime = System.currentTimeMillis();
            System.out.printf("导入完成: %d条记录, 耗时: %dms%n", totalRecords, endTime - startTime);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
        }
    }
    private void readFileInBatches(FileChannel channel, BlockingQueue<List<DataRecord>> queue) throws IOException {
        // 使用内存映射文件提高读取速度
        MappedByteBuffer buffer = channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size());
        String content = StandardCharsets.UTF_8.decode(buffer).toString();
        // 使用并行流处理字符串
        List<DataRecord> batch = new ArrayList<>(BATCH_SIZE);
        int count = 0;
        for (String line : content.split("\n")) {
            String[] fields = line.split(",", -1); // 保留空字段
            // 使用StringBuilder优化字符串拼接
            DataRecord record = parseRecord(fields);
            batch.add(record);
            if (++count % BATCH_SIZE == 0) {
                try {
                    queue.put(new ArrayList<>(batch));
                    batch.clear();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
        // 处理剩余数据
        if (!batch.isEmpty()) {
            try {
                queue.put(batch);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    private DataRecord parseRecord(String[] fields) {
        DataRecord record = new DataRecord();
        // 使用预编译的正则表达式
        record.setId(parseLongFast(fields[0]));
        record.setName(fields[1].trim());
        record.setAge(parseIntFast(fields[2]));
        return record;
    }
    // 自定义快速解析方法
    private long parseLongFast(String str) {
        long result = 0;
        boolean negative = false;
        int i = 0, len = str.length();
        if (len > 0) {
            if (str.charAt(0) == '-') {
                negative = true;
                i = 1;
            }
            while (i < len) {
                result = result * 10 + (str.charAt(i++) - '0');
            }
        }
        return negative ? -result : result;
    }
    private int parseIntFast(String str) {
        return (int) parseLongFast(str);
    }
    // 数据库写入线程
    static class DatabaseWriter implements Callable<Integer> {
        private final BlockingQueue<List<DataRecord>> queue;
        private final ConnectionPool connectionPool;
        DatabaseWriter(BlockingQueue<List<DataRecord>> queue) {
            this.queue = queue;
            this.connectionPool = ConnectionPool.getInstance();
        }
        @Override
        public Integer call() {
            int count = 0;
            String sql = "INSERT INTO records (id, name, age) VALUES (?, ?, ?)";
            try (Connection conn = connectionPool.getConnection();
                 PreparedStatement ps = conn.prepareStatement(sql)) {
                while (true) {
                    List<DataRecord> batch = queue.take();
                    if (batch.isEmpty()) break;
                    // 批量处理
                    for (DataRecord record : batch) {
                        ps.setLong(1, record.getId());
                        ps.setString(2, record.getName());
                        ps.setInt(3, record.getAge());
                        ps.addBatch();
                    }
                    int[] results = ps.executeBatch();
                    for (int result : results) {
                        count += result;
                    }
                    // 定期提交事务
                    if (count % 10000 == 0) {
                        conn.commit();
                    }
                }
                conn.commit();
            } catch (Exception e) {
                e.printStackTrace();
            }
            return count;
        }
    }
}

Excel文件导入优化案例

优化前：使用POI默认方式

public class SlowExcelImport {
    public void importExcel(String filePath) {
        try (Workbook workbook = WorkbookFactory.create(new File(filePath))) {
            Sheet sheet = workbook.getSheetAt(0);
            // 问题：逐行逐单元格处理，效率低下
            for (Row row : sheet) {
                for (Cell cell : row) {
                    String value = getCellValue(cell);
                    // 处理每个单元格
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private String getCellValue(Cell cell) {
        // 大量类型判断，效率低下
        switch (cell.getCellType()) {
            case STRING: return cell.getStringCellValue();
            case NUMERIC: return String.valueOf(cell.getNumericCellValue());
            case BOOLEAN: return String.valueOf(cell.getBooleanCellValue());
            default: return "";
        }
    }
}

优化后：使用事件驱动模式

public class OptimizedExcelImport {
    public void importLargeExcel(String filePath) {
        try {
            long startTime = System.currentTimeMillis();
            // 使用SAX方式解析大文件
            OPCPackage pkg = OPCPackage.open(new File(filePath));
            XSSFReader reader = new XSSFReader(pkg);
            StylesTable styles = reader.getStylesTable();
            // 获取共享字符串表
            ReadOnlySharedStringsTable strings = new ReadOnlySharedStringsTable(pkg);
            // 创建SAX解析器
            XMLReader parser = XMLReaderFactory.createXMLReader();
            // 设置内容处理器
            SheetHandler handler = new SheetHandler(strings);
            parser.setContentHandler(handler);
            // 逐sheet解析
            Iterator<InputStream> sheets = reader.getSheetsData();
            while (sheets.hasNext()) {
                InputStream sheet = sheets.next();
                parser.parse(new InputSource(sheet));
                sheet.close();
            }
            long endTime = System.currentTimeMillis();
            System.out.println("导入耗时: " + (endTime - startTime) + "ms");
            pkg.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    // 使用SAX事件处理，避免加载整个文档
    static class SheetHandler extends DefaultHandler {
        private final ReadOnlySharedStringsTable strings;
        private StringBuilder currentValue;
        private List<String> rowValues;
        private List<List<String>> rows;
        private boolean isInCell;
        SheetHandler(ReadOnlySharedStringsTable strings) {
            this.strings = strings;
            this.currentValue = new StringBuilder();
            this.rowValues = new ArrayList<>();
            this.rows = new ArrayList<>();
        }
        @Override
        public void startElement(String uri, String localName, String qName, Attributes attrs) {
            if ("c".equals(qName)) {
                // 开始单元格
                isInCell = true;
                currentValue.setLength(0);
            } else if ("row".equals(qName)) {
                // 开始行
                rowValues.clear();
            } else if ("t".equals(qName)) {
                // 包含字符串类型标记
            }
        }
        @Override
        public void characters(char[] ch, int start, int length) {
            if (isInCell) {
                currentValue.append(ch, start, length);
            }
        }
        @Override
        public void endElement(String uri, String localName, String qName) {
            if ("c".equals(qName)) {
                // 结束单元格
                rowValues.add(currentValue.toString());
                isInCell = false;
            } else if ("row".equals(qName)) {
                // 结束行，处理该行数据
                processRow(new ArrayList<>(rowValues));
            }
        }
        private void processRow(List<String> rowData) {
            // 批量处理行数据
            if (!rowData.isEmpty()) {
                // 可以在这里进行批量插入或其他处理
                rows.add(rowData);
                if (rows.size() >= 1000) {
                    // 批量写入数据库
                    batchInsert(rows);
                    rows.clear();
                }
            }
        }
        private void batchInsert(List<List<String>> rows) {
            // 批量插入逻辑
            String sql = "INSERT INTO excel_records VALUES (?, ?, ?)";
            try (Connection conn = getConnection();
                 PreparedStatement ps = conn.prepareStatement(sql)) {
                conn.setAutoCommit(false);
                for (List<String> row : rows) {
                    for (int i = 0; i < row.size(); i++) {
                        ps.setString(i + 1, row.get(i));
                    }
                    ps.addBatch();
                }
                ps.executeBatch();
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        private Connection getConnection() {
            // 数据库连接逻辑
            return null;
        }
    }
}

JSON文件导入优化案例

优化前：使用Jackson默认方式

public class SlowJSONImport {
    public void importJSON(String filePath) {
        ObjectMapper mapper = new ObjectMapper();
        try {
            // 问题：一次性加载整个JSON到内存
            JsonNode root = mapper.readTree(new File(filePath));
            // 问题：递归遍历，性能低下
            processNode(root);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private void processNode(JsonNode node) {
        // 递归处理，大量对象创建
        if (node.isObject()) {
            Iterator<String> fieldNames = node.fieldNames();
            while (fieldNames.hasNext()) {
                String fieldName = fieldNames.next();
                processNode(node.get(fieldName));
            }
        } else if (node.isArray()) {
            for (JsonNode child : node) {
                processNode(child);
            }
        }
    }
}

优化后：使用流式解析

public class OptimizedJSONImport {
    public void importLargeJSON(String filePath) {
        try {
            long startTime = System.currentTimeMillis();
            // 使用Jackson的流式API
            JsonFactory factory = new JsonFactory();
            try (JsonParser parser = factory.createParser(new File(filePath))) {
                // 跳过开始的[
                parser.nextToken();
                // 使用BFS方式解析
                Queue<JsonRecord> recordQueue = new LinkedList<>();
                List<JsonRecord> batch = new ArrayList<>();
                while (parser.nextToken() != JsonToken.END_ARRAY) {
                    JsonRecord record = parseRecord(parser);
                    recordQueue.offer(record);
                    // 批量处理
                    if (recordQueue.size() >= 1000) {
                        batchInsert(recordQueue);
                        recordQueue.clear();
                    }
                }
                // 处理剩余数据
                if (!recordQueue.isEmpty()) {
                    batchInsert(recordQueue);
                }
            }
            long endTime = System.currentTimeMillis();
            System.out.println("导入耗时: " + (endTime - startTime) + "ms");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private JsonRecord parseRecord(JsonParser parser) throws IOException {
        JsonRecord record = new JsonRecord();
        // 开始解析对象
        parser.nextToken(); // 跳过{
        while (parser.nextToken() != JsonToken.END_OBJECT) {
            String fieldName = parser.getCurrentName();
            parser.nextToken();
            switch (fieldName) {
                case "id":
                    record.setId(parser.getLongValue());
                    break;
                case "name":
                    record.setName(parser.getText());
                    break;
                case "age":
                    record.setAge(parser.getIntValue());
                    break;
                // 其他字段处理
                default:
                    parser.skipChildren();
            }
        }
        return record;
    }
    private void batchInsert(Queue<JsonRecord> records) {
        // 批量插入逻辑，使用连接池和批量操作
        try (Connection conn = getConnection();
             PreparedStatement ps = conn.prepareStatement(
                 "INSERT INTO json_records (id, name, age) VALUES (?, ?, ?)")) {
            conn.setAutoCommit(false);
            while (!records.isEmpty()) {
                JsonRecord record = records.poll();
                ps.setLong(1, record.getId());
                ps.setString(2, record.getName());
                ps.setInt(3, record.getAge());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
    private Connection getConnection() {
        // 使用HikariCP连接池
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mysql://localhost:3306/test");
        config.setUsername("user");
        config.setPassword("pass");
        config.setMaximumPoolSize(10);
        config.setConnectionTimeout(30000);
        HikariDataSource dataSource = new HikariDataSource(config);
        try {
            return dataSource.getConnection();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}

通用优化策略总结

public class ImportOptimizationStrategies {
    // 1. 使用内存映射文件
    public MappedByteBuffer useMemoryMappedFile(String filePath) throws IOException {
        FileChannel channel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ);
        return channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size());
    }
    // 2. 使用连接池
    public class ConnectionPool {
        private static final HikariDataSource dataSource;
        static {
            HikariConfig config = new HikariConfig();
            config.setMaximumPoolSize(20);
            config.setMinimumIdle(5);
            config.setIdleTimeout(300000);
            config.setConnectionTimeout(10000);
            config.setMaxLifetime(1200000);
            dataSource = new HikariDataSource(config);
        }
        public static Connection getConnection() throws SQLException {
            return dataSource.getConnection();
        }
    }
    // 3. 使用批量操作
    public int batchInsert(List<DataRecord> records) {
        String sql = "INSERT INTO records (col1, col2, col3) VALUES (?, ?, ?)";
        try (Connection conn = ConnectionPool.getConnection();
             PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (int i = 0; i < records.size(); i += 1000) {
                int end = Math.min(i + 1000, records.size());
                List<DataRecord> batch = records.subList(i, end);
                for (DataRecord record : batch) {
                    ps.setObject(1, record.getCol1());
                    ps.setObject(2, record.getCol2());
                    ps.setObject(3, record.getCol3());
                    ps.addBatch();
                }
                ps.executeBatch();
                conn.commit();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return records.size();
    }
    // 4. 使用并行处理
    public void parallelProcessing(List<DataRecord> records) {
        int processors = Runtime.getRuntime().availableProcessors();
        ExecutorService executor = Executors.newFixedThreadPool(processors);
        int batchSize = records.size() / processors;
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 0; i < processors; i++) {
            int start = i * batchSize;
            int end = (i == processors - 1) ? records.size() : (i + 1) * batchSize;
            List<DataRecord> subList = records.subList(start, end);
            futures.add(executor.submit(() -> batchInsert(subList)));
        }
        // 等待所有任务完成
        int total = 0;
        for (Future<Integer> future : futures) {
            try {
                total += future.get();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        executor.shutdown();
    }
    // 5. 使用零拷贝技术
    public void zeroCopyTransfer(String srcPath, String destPath) throws IOException {
        try (FileChannel src = FileChannel.open(Paths.get(srcPath), StandardOpenOption.READ);
             FileChannel dest = FileChannel.open(Paths.get(destPath), 
                 StandardOpenOption.WRITE, StandardOpenOption.CREATE)) {
            long position = 0;
            long size = src.size();
            while (position < size) {
                position += src.transferTo(position, size - position, dest);
            }
        }
    }
    // 6. 使用缓存和索引
    public class CachedImport {
        private final Map<String, Integer> cache = new HashMap<>(100000);
        public void importWithCache(List<DataRecord> records) {
            records.parallelStream().forEach(record -> {
                // 使用缓存避免重复计算
                cache.computeIfAbsent(record.getKey(), k -> {
                    // 只有首次命中才进行计算
                    return expensiveCalculation(record);
                });
            });
        }
        private int expensiveCalculation(DataRecord record) {
            // 复杂的计算逻辑
            return 0;
        }
    }
}

性能对比表

最佳实践建议

1. 文件分割：大文件先分割成小文件再处理
2. 内存控制：合理设置JVM参数，使用-Xmx控制堆内存
3. GC优化：使用G1GC或ZGC减少停顿时间
4. 监测工具：使用VisualVM或JProfiler监控性能瓶颈
5. 异步处理：使用消息队列实现异步导入
6. 断点续传：记录处理进度，支持失败重试