多线程读取并解析csv

import java.util.*; /** * 1. 定义提取策略：封装“如何从原始行提取数据”的逻辑 */ class ColumnExtractionStrategy { private final int[] indices; public ColumnExtractionStrategy(int... indices) { this.indices = indices; } /** * 执行提取 */ public List<String> extract(String[] rawColumns) { List<String> result = new ArrayList<>(); for (int index : indices) { if (index < rawColumns.length) { result.add(rawColumns[index].trim()); // 自动去空格 } else { result.add(""); // 防止数组越界，给空值 } } return result; } } /** * 2. 策略注册表：静态管理所有文件的提取规则 */ public class StrategyConfig { // 注册表：Key是文件名特征，Value是提取策略 private static final Map<String, ColumnExtractionStrategy> STRATEGY_MAP = new HashMap<>(); static { // 类3 CSV：提取第 3, 4, 5, 6 列 (对应数组下标) STRATEGY_MAP.put("type3", new ColumnExtractionStrategy(3, 4, 5, 6)); // 类4 CSV：提取第 2, 4 列 STRATEGY_MAP.put("type4", new ColumnExtractionStrategy(2, 4)); // 类5 CSV：提取第 0, 1 列 (假设是经纬度) STRATEGY_MAP.put("type5", new ColumnExtractionStrategy(0, 1)); } /** * 根据文件名获取策略 */ public static ColumnExtractionStrategy getStrategy(String fileName) { for (Map.Entry<String, ColumnExtractionStrategy> entry : STRATEGY_MAP.entrySet()) { if (fileName.contains(entry.getKey())) { System.out.println("匹配到策略: " + entry.getKey()); return entry.getValue(); } } return null; } }

import java.util.Arrays; /** * 3. 通用行数据对象 */ class RawRow { private long rowIndex; private String[] columns; public RawRow(long rowIndex, String[] columns) { this.rowIndex = rowIndex; this.columns = columns; } public long getRowIndex() { return rowIndex; } public String[] getColumns() { return columns; } }

import java.io.BufferedReader; import java.io.FileReader; import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; /** * 4. 生产者：负责读取 CSV 并利用策略过滤列 */ public class CsvProducer { private final BlockingQueue<List<RawRow>> queue; private static final int BATCH_SIZE = 1000; public CsvProducer(BlockingQueue<List<RawRow>> queue) { this.queue = queue; } /** * 读取文件 * @param filePath 文件路径 * @param strategy 提取策略（如果为 null，则不提取，直接透传所有列） */ public void produce(String filePath, ColumnExtractionStrategy strategy) throws IOException { try (BufferedReader br = new BufferedReader(new FileReader(filePath))) { // 跳过表头 String header = br.readLine(); if (header == null) return; String line; long rowIndex = 0; List<RawRow> buffer = new ArrayList<>(); while ((line = br.readLine()) != null) { rowIndex++; String[] rawCols = line.split(","); // 简单分割 // === 核心：利用策略过滤数据 === String[] finalCols; if (strategy != null) { // 如果找到了策略，只提取需要的列 List<String> extracted = strategy.extract(rawCols); finalCols = extracted.toArray(new String[0]); } else { // 如果没有策略（普通文件），使用原始列 finalCols = rawCols; } buffer.add(new RawRow(rowIndex, finalCols)); // 批量入队 if (buffer.size() >= BATCH_SIZE) { queue.put(new ArrayList<>(buffer)); buffer.clear(); } } // 剩余数据入队 if (!buffer.isEmpty()) { queue.put(new ArrayList<>(buffer)); } } catch (InterruptedException e) { e.printStackTrace(); } } }

import java.util.*; import java.util.concurrent.BlockingQueue; /** * 5. 消费者：从队列取数据，组装成 Map (模拟 MyBatis 入库) */ public class MapBasedConsumer implements Runnable { private final BlockingQueue<List<RawRow>> queue; private final List<RawRow> POISON_PILL = new ArrayList<>(); private final int threadId; // 假设这是数据库字段名（实际项目中可以从配置里读，或者根据下标硬编码） // 注意：这里的顺序必须和 StrategyConfig 里的列顺序对应！ // 比如 type3 是 [3,4,5,6]，那这里就是 [fieldA, fieldB, fieldC, fieldD] private final List<String> targetFields; public MapBasedConsumer(BlockingQueue<List<RawRow>> queue, int threadId, List<String> targetFields) { this.queue = queue; this.threadId = threadId; this.targetFields = targetFields; } @Override public void run() { try { while (true) { List<RawRow> batch = queue.take(); if (batch == POISON_PILL) break; for (RawRow row : batch) { // === 组装 Map === Map<String, Object> dataMap = new HashMap<>(); String[] cols = row.getColumns(); for (int i = 0; i < cols.length; i++) { // 防止配置不对导致数组越界 if (i < targetFields.size()) { dataMap.put(targetFields.get(i), cols[i]); } } // === 模拟 MyBatis 插入 === // System.out.println("线程 " + threadId + " 准备插入: " + dataMap); // genericMapper.insert(dataMap); } System.out.println("线程 " + threadId + " 处理了一批: " + batch.size()); } } catch (Exception e) { e.printStackTrace(); } } }

import java.util.Arrays; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * 6. 主程序入口 */ public class Main { // 共享队列 private static final BlockingQueue<List<RawRow>> queue = new ArrayBlockingQueue<>(10); private static final List<RawRow> POISON_PILL = new ArrayList<>(); public static void main(String[] args) { String filePath = "data_type3.csv"; // 测试文件名 // 1. 根据文件名获取策略 ColumnExtractionStrategy strategy = StrategyConfig.getStrategy(filePath); // 2. 启动消费者 // 注意：这里需要告诉消费者目标字段名，以便组装 Map // 实际项目中，这个字段列表也可以放在 StrategyConfig 里，或者通过注解获取 // 这里为了演示，假设 type3 对应的数据库字段是 name, age, city, phone java.util.List<String> dbFields = Arrays.asList("name", "age", "city", "phone"); ExecutorService executor = Executors.newFixedThreadPool(4); for (int i = 0; i < 4; i++) { executor.submit(new MapBasedConsumer(queue, i, dbFields)); } // 3. 启动生产者 CsvProducer producer = new CsvProducer(queue); try { producer.produce(filePath, strategy); } catch (Exception e) { e.printStackTrace(); } // 4. 发送结束信号 for (int i = 0; i < 4; i++) { try { queue.put(POISON_PILL); } catch (InterruptedException e) {} } executor.shutdown(); System.out.println("程序执行完毕"); } }