如何使用Spring Batch读取行相互依赖的固定长度扁平文件

算力阿峰

2026-5-25

嘿，这个带层级的固定长度文件处理场景在Spring Batch里太常见了！我来给你梳理一套完整的实现方案，从模型定义到Job配置一步到位：

1. 先定义对应的领域模型

首先得把四种行类型映射成Java类，每个类的字段严格对应文件里的固定长度位置：

// 头行（00开头）
public class FileHeader {
    @Field(pos = 1, length = 2)
    private String recordType; // 固定值"00"
    @Field(pos = 3, length = 10)
    private Integer dataRecordCount; // 记录10类型数据行的总数量
    // 其他固定长度的业务字段，按实际位置定义
    // getter/setter省略
}

// 主数据行（10开头）
public class DataRecord {
    @Field(pos = 1, length = 2)
    private String recordType; // 固定值"10"
    @Field(pos = 3, length = 15)
    private String dataId; // 唯一标识，用来关联子数据行
    @Field(pos = 18, length = 5)
    private Integer childRecordCount; // 对应关联的20类型子行数量
    // 其他业务字段
    // getter/setter省略
}

// 子数据行（20开头）
public class ChildDataRecord {
    @Field(pos = 1, length = 2)
    private String recordType; // 固定值"20"
    @Field(pos = 3, length = 15)
    private String dataId; // 关联主数据行的dataId
    // 其他子业务字段
    // getter/setter省略
}

// 尾行（99开头）
public class FileFooter {
    @Field(pos = 1, length = 2)
    private String recordType; // 固定值"99"
    @Field(pos = 3, length = 10)
    private Integer dataRecordCount; // 需和头行的数量一致
    // 其他字段
    // getter/setter省略
}

// 聚合对象：把主数据行和对应的子行集合封装在一起，方便后续处理
public class AggregatedData {
    private DataRecord parent;
    private List<ChildDataRecord> children;
    // getter/setter/构造方法省略
}

2. 配置FlatFileItemReader，区分不同行类型

核心是用PatternMatchingCompositeLineTokenizer根据前两位字符匹配不同的解析规则，再结合BeanWrapperFieldSetMapper把解析出的字段映射到对应的模型类：

@Configuration
public class BatchReaderConfig {

    @Bean
    public FlatFileItemReader<Object> fixedLengthFileReader() {
        FlatFileItemReader<Object> reader = new FlatFileItemReader<>();
        reader.setResource(new ClassPathResource("your-data-file.txt")); // 替换成你的文件路径
        reader.setLineMapper(lineMapper());
        reader.setEncoding("UTF-8"); // 根据文件实际编码调整，比如GBK
        return reader;
    }

    private LineMapper<Object> lineMapper() {
        DefaultLineMapper<Object> lineMapper = new DefaultLineMapper<>();
        lineMapper.setLineTokenizer(compositeLineTokenizer());
        lineMapper.setFieldSetMapper(compositeFieldSetMapper());
        return lineMapper;
    }

    private PatternMatchingCompositeLineTokenizer compositeLineTokenizer() {
        PatternMatchingCompositeLineTokenizer tokenizer = new PatternMatchingCompositeLineTokenizer();

        Map<String, LineTokenizer> tokenizers = new HashMap<>();
        tokenizers.put("00*", headerLineTokenizer());
        tokenizers.put("10*", dataLineTokenizer());
        tokenizers.put("20*", childLineTokenizer());
        tokenizers.put("99*", footerLineTokenizer());

        tokenizer.setTokenizers(tokenizers);
        return tokenizer;
    }

    // 头行的固定长度Tokenizer（Spring Batch的Range从1开始计数）
    private LineTokenizer headerLineTokenizer() {
        FixedLengthTokenizer tokenizer = new FixedLengthTokenizer();
        tokenizer.setNames("recordType", "dataRecordCount");
        tokenizer.setColumns(new Range(1,2), new Range(3,12));
        tokenizer.setStrict(true); // 严格校验行长度，避免解析错误
        return tokenizer;
    }

    // 主数据行的Tokenizer
    private LineTokenizer dataLineTokenizer() {
        FixedLengthTokenizer tokenizer = new FixedLengthTokenizer();
        tokenizer.setNames("recordType", "dataId", "childRecordCount");
        tokenizer.setColumns(new Range(1,2), new Range(3,17), new Range(18,22));
        tokenizer.setStrict(true);
        return tokenizer;
    }

    // 子数据行、尾行的Tokenizer同理，按实际字段位置配置即可，这里省略

    private CompositeFieldSetMapper<Object> compositeFieldSetMapper() {
        CompositeFieldSetMapper<Object> fieldSetMapper = new CompositeFieldSetMapper<>();

        Map<String, FieldSetMapper<?>> mappers = new HashMap<>();
        mappers.put("00*", new BeanWrapperFieldSetMapper<>() {{
            setTargetType(FileHeader.class);
        }});
        mappers.put("10*", new BeanWrapperFieldSetMapper<>() {{
            setTargetType(DataRecord.class);
        }});
        mappers.put("20*", new BeanWrapperFieldSetMapper<>() {{
            setTargetType(ChildDataRecord.class);
        }});
        mappers.put("99*", new BeanWrapperFieldSetMapper<>() {{
            setTargetType(FileFooter.class);
        }});

        fieldSetMapper.setFieldSetMappers(mappers);
        return fieldSetMapper;
    }
}

3. 实现ItemProcessor聚合父子数据

因为文件里主数据行（10）后面直接跟着对应的子数据行（20），我们可以在Processor里维护临时缓存，收集子行直到达到主行指定的数量，再输出聚合后的对象：

@Component
public class DataAggregationProcessor implements ItemProcessor<Object, AggregatedData> {

    private DataRecord currentParent;
    private List<ChildDataRecord> currentChildren = new ArrayList<>();
    private FileHeader fileHeader;
    private int processedDataRecords = 0;

    @Override
    public AggregatedData process(Object item) throws Exception {
        if (item instanceof FileHeader) {
            this.fileHeader = (FileHeader) item;
            return null; // 头行暂不输出，留到末尾做校验
        } else if (item instanceof DataRecord) {
            // 如果有未处理完的主行，先输出之前的聚合结果
            if (currentParent != null) {
                AggregatedData aggregated = new AggregatedData(currentParent, currentChildren);
                currentParent = (DataRecord) item;
                currentChildren.clear();
                processedDataRecords++;
                return aggregated;
            }
            currentParent = (DataRecord) item;
            currentChildren.clear();
            return null;
        } else if (item instanceof ChildDataRecord) {
            currentChildren.add((ChildDataRecord) item);
            // 检查是否已收集到指定数量的子行
            if (currentChildren.size() == currentParent.getChildRecordCount()) {
                AggregatedData aggregated = new AggregatedData(currentParent, currentChildren);
                currentParent = null;
                currentChildren.clear();
                processedDataRecords++;
                return aggregated;
            }
            return null;
        } else if (item instanceof FileFooter) {
            FileFooter footer = (FileFooter) item;
            // 校验头尾行的数量是否一致，且和实际处理的主行数量匹配
            if (!fileHeader.getDataRecordCount().equals(footer.getDataRecordCount()) 
                || !fileHeader.getDataRecordCount().equals(processedDataRecords)) {
                throw new JobExecutionException("数据行数量不匹配！头行记录：" + fileHeader.getDataRecordCount() 
                                                + "，尾行记录：" + footer.getDataRecordCount() 
                                                + "，实际处理：" + processedDataRecords);
            }
            return null; // 尾行只做校验，不输出
        }
        return null;
    }
}

4. 配置Job和Step

最后把Reader、Processor和自定义的Writer（比如写入数据库、生成报表等）组装成Step和Job：

@Configuration
@EnableBatchProcessing
public class BatchJobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private FlatFileItemReader<Object> fixedLengthFileReader;
    @Autowired
    private DataAggregationProcessor aggregationProcessor;
    @Autowired
    private ItemWriter<AggregatedData> aggregatedDataWriter; // 你需要根据业务实现这个Writer

    @Bean
    public Step dataProcessingStep() {
        return stepBuilderFactory.get("dataProcessingStep")
                .<Object, AggregatedData>chunk(100) // 根据文件大小调整chunk大小
                .reader(fixedLengthFileReader)
                .processor(aggregationProcessor)
                .writer(aggregatedDataWriter)
                .build();
    }

    @Bean
    public Job fixedLengthFileProcessingJob() {
        return jobBuilderFactory.get("fixedLengthFileProcessingJob")
                .start(dataProcessingStep())
                .build();
    }
}

一些关键注意事项

字段转换：如果有数字、日期类型的字段，记得给BeanWrapperFieldSetMapper配置ConversionService，处理带空格填充的数字（比如" 1234"转成Integer）或者自定义日期格式。
异常处理：可以在Step里配置SkipPolicy，允许跳过个别格式错误的行，避免整个Job失败。
性能优化：如果文件特别大，可以调整chunk大小，或者用MultiResourceItemReader处理多文件批量读取。

内容的提问来源于stack exchange，提问作者thierryler