spring batch 纯注解学习笔记(五)--文件读写

作者: 小偷阿辉 | 来源:发表于2020-12-31 08:39 被阅读0次

在Spring batch由上至下的结构中Job、Step都是属于框架级别的的功能，大部分时候都是提供一些配置选项给开发人员使用，而Item中的Reader、Processor和Writer是属于业务级别的，它开放了一些业务切入的接口。但是文件的读写过程中有很多通用一致的功能Spring Batch为这些相同的功能提供了一致性实现类。

1.扁平结构文件

扁平结构文件（也称为矩阵结构文件，后文简称为文件）是最常见的一种文件类型。他通常以一行表示一条记录，字段数据之间用某种方式分割。与标准的格式数据（xml、json等）主要差别在于他没有结构性描述方案（SXD、JSON-SCHEME），进而没有结构性分割规范。因此在读写此类文件之前需要先设定好字段的分割方法。

文件的字段数据分割方式通常有两种：使用分隔符或固定字段长度。前者通常使用逗号（，）之类的符号对字段数据进行划分，后者的每一列字段数据长度是固定的。框架为文件的读取提供了FieldSet用于将文件结构中的信息映射到一个对象。FieldSet的作用是将文件的数据与类的field进行绑定（field是Java中常见的概念，不清楚的可以了解Java反射）。

2.数据读取

pring Batch为文件读取提供了FlatFileItemReader类，它为文件中的数据的读取和转换提供了基本功能。在FlatFileItemReader中有2个主要的功能接口，一是Resource、二是LineMapper。 Resource用于外部文件获取，下面是一个例子：

Resource resource = new FileSystemResource("resources/trades.csv");

在复杂的生产环境中，文件通常由中心化、或者流程式的基础框架来管理（比如EAI）。因此文件往往需要使用FTP等方式从其他位置获取。如何迁移文件已经超出了Spring Batch框架的范围，在Spring的体系中可以参考Spring Integration项目。

下面是FlatFileItemReader的属性，每一个属性都提供了Setter方法。

属性

每个属性都为文件的解析提供了某方面的功能，下面是结构的说明。

2.1.LineMapper

这个接口的作用是将字符串转换为对象：

public interface LineMapper { T mapLine(String line, int lineNumber) throws Exception; }

接口的基本处理逻辑是聚合类（FlatFileItemReader）传递一行字符串以及行号给LineMapper::mapLine，方法处理后返回一个映射的对象。

2.1.LineTokenizer

这个接口的作用是将一行数据转换为一个FieldSet结构。对于Spring Batch而言，扁平结构文件的到Java实体的映射都通过FieldSet来控制，因此读写文件的过程需要完成字符串到FieldSet的转换：

public interface LineTokenizer { FieldSet tokenize(String line); }

这个接口的含义是：传递一行字符串数据，然后获取一个FieldSet。

框架为LineTokenizer提供三个实现类：

DelimitedLineTokenizer：利用分隔符将数据转换为FieldSet。最常见的分隔符是逗号,，类提供了分隔符的配置和解析方法。
FixedLengthTokenizer：根据字段的长度来解析出FieldSet结构。必须为记录定义字段宽度。
PatternMatchingCompositeLineTokenizer：使用一个匹配机制来动态决定使用哪个LineTokenizer。

2.3.FieldSetMapper

该接口是将FieldSet转换为对象：

public interface FieldSetMapper { T mapFieldSet(FieldSet fieldSet) throws BindException; }

FieldSetMapper通常和LineTokenizer联合在一起使用：String->FieldSet->Object。

2.4.DefaultLineMapper

DefaultLineMapper是LineMapper的实现，他实现了从文件到Java实体的映射：

public class DefaultLineMapper implements LineMapper<>, InitializingBean {
    private LineTokenizer tokenizer;
    private FieldSetMapper fieldSetMapper;
    public T mapLine(String line, int lineNumber) throws Exception {
        return fieldSetMapper.mapFieldSet(tokenizer.tokenize(line));
    }
    public void setLineTokenizer(LineTokenizer tokenizer) {
        this.tokenizer = tokenizer;
    }
    public void setFieldSetMapper(FieldSetMapper fieldSetMapper) {
        this.fieldSetMapper = fieldSetMapper;
    }
}

在解析文件时数据是按行解析的：

传入一行字符串。
LineTokenizer将字符串解析为FieldSet结构。
FieldSetMapper继续解析为一个Java实体对象返回给调用者。

DefaultLineMapper是框架提供的默认实现类，看似非常简单，但是利用组合模式可以扩展出很多功能。

2.5.数据自动映射

在转换过程中如果将FieldSet的names属性与目标类的field绑定在一起，那么可以直接使用反射实现数据转换，为此框架提供了BeanWrapperFieldSetMapper来实现。

DefaultLineMapper<WeatherEntity> lineMapper = new DefaultLineMapper<>(); //创建LineMapper

DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer(); //创建LineTokenizer
tokenizer.setNames(new String[] { "siteId", "month", "type", "value", "ext" }); //设置Field名称

BeanWrapperFieldSetMapper<WeatherEntity> wrapperMapper 
    = new BeanWrapperFieldSetMapper<>(); //创建FieldSetMapper
wrapperMapper.setTargetType(WeatherEntity.class); //设置实体，实体的field名称必须和tokenizer.names一致。

// 组合lineMapper
lineMapper.setLineTokenizer(tokenizer);
lineMapper.setFieldSetMapper(wrapperMapper);

2.6.文件读取总结

上面提到了各种接口和实现，实际上都是围绕着FlatFileItemReader的属性在介绍，虽然内容很多但是实际上就以下几点：

首先要定位文件，Spring Batch提供了Resource相关的定位方法。
其次是将文件中的行字符串数据转换为对象，LineMapper的功能就是完成这个功能。
框架为LineMapper提供了DefaultLineMapper作为默认实现方法，在DefaultLineMapper中需要组合使用LineTokenizer和FieldSetMapper。前者将字符串转为为一个Field，后者将Field转换为目标对象。
LineTokenizer有3个实现类可供使用、FieldSetMapper有一个默认实现类BeanWrapperFieldSetMapper。

2.7.文件读写可执行源码

运行之前需要配置数据库链接，参看源码库中的README.md。

文件读取的主要逻辑在org.chenkui.spring.batch.sample.items.FlatFileReader类：

public class FlatFileReader {
    // FeildSet的字段名，设置字段名之后可以直接使用名字作为索引获取数据。也可以使用索引位置来获取数据
    public final static String[] Tokenizer = new String[] { "siteId", "month", "type", "value", "ext" };
    private boolean userWrapper = false;

    @Bean
    //定义FieldSetMapper用于FieldSet->WeatherEntity
    public FieldSetMapper<WeatherEntity> fieldSetMapper() {
        return new FieldSetMapper<WeatherEntity>() {
            @Override
            public WeatherEntity mapFieldSet(FieldSet fieldSet) throws BindException {
                if (null == fieldSet) {
                    return null; // fieldSet不存在则跳过该行处理
                } else {
                    WeatherEntity observe = new WeatherEntity();
                    observe.setSiteId(fieldSet.readRawString("siteId"));
                    //Setter
                    return observe;
                }
            }
        };
    }

    @Bean
    // 配置 Reader
    public ItemReader<WeatherEntity> flatFileReader(
                           @Qualifier("fieldSetMapper") FieldSetMapper<WeatherEntity> fieldSetMapper) {
        FlatFileItemReader<WeatherEntity> reader = new FlatFileItemReader<>();
        reader.setResource(new FileSystemResource("src/main/resources/data.csv")); // 读取资源文件
        DefaultLineMapper<WeatherEntity> lineMapper = new DefaultLineMapper<>(); // 初始化 LineMapper实现类
        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer(); // 创建LineTokenizer接口实现

        tokenizer.setNames(Tokenizer); // 设定每个字段的名称，如果不设置需要使用索引获取值
        lineMapper.setLineTokenizer(tokenizer); // 设置tokenizer工具

        if (userWrapper) { //使用 BeanWrapperFieldSetMapper 使用反射直接转换
            BeanWrapperFieldSetMapper<WeatherEntity> wrapperMapper = new BeanWrapperFieldSetMapper<>();
            wrapperMapper.setTargetType(WeatherEntity.class);
            fieldSetMapper = wrapperMapper;
        }

        lineMapper.setFieldSetMapper(fieldSetMapper);
        reader.setLineMapper(lineMapper);
        reader.setLinesToSkip(1); // 跳过的初始行，用于过滤字段行
        reader.open(new ExecutionContext());
        return reader;
    }
}

2.8.按字段长度格读取文件

除了按照分隔符，有些文件可以字段数据的占位长度来提取数据。按照前面介绍的过程，实际上只要修改LineTokenizer接口即可，框架提供了FixedLengthTokenizer类：

@Bean
public FixedLengthTokenizer fixedLengthTokenizer() {
    FixedLengthTokenizer tokenizer = new FixedLengthTokenizer();

    tokenizer.setNames("ISIN", "Quantity", "Price", "Customer");
    //Range用于设定数据的长度。
    tokenizer.setColumns(new Range(1-12),
                        new Range(13-15),
                        new Range(16-20),
                        new Range(21-29));
    return tokenizer;
}

3.0写入扁平结构文件

将数据写入到文件与读取的过程正好相反：将对象转换为字符串。

3.1.LineAggregator

与LineMapper相对应的是LineAggregator，他的功能是将实体转换为字符串：

public interface LineAggregator<T> {
    public String aggregate(T item);
}

3.2.PassThroughLineAggregator

框架为LineAggregator接口提供了一个非常简单的实现类——PassThroughLineAggregator，其唯一实现就是使用对象的toString方法：

3.3.DelimitedLineAggregator

ineAggregator的另外一个实现类是DelimitedLineAggregator。与PassThroughLineAggregator简单直接使用toString方法不同的是，DelimitedLineAggregator需要一个转换接口FieldExtractor：

DelimitedLineAggregator<CustomerCredit> lineAggregator = new DelimitedLineAggregator<>();
lineAggregator.setDelimiter(",");
lineAggregator.setFieldExtractor(fieldExtractor);

3.4.FieldExtractor

FieldExtractor用于实体类到collection结构的转换。它可以和LineTokenizer进行类比，前者是将实体类转换为扁平结构的数据，后者是将String转换为一个FieldSet结构。

public interface FieldExtractor<T> {
    Object[] extract(T item);
}

框架为FieldExtractor接口提供了一个基于反射的实现类BeanWrapperFieldExtractor，其过程就是将实体对象转换为列表：

BeanWrapperFieldExtractor<CustomerCredit> fieldExtractor = new BeanWrapperFieldExtractor<>();
fieldExtractor.setNames(new String[] {"field1", "field2"});

setName方法用于指定要转换的field列表。

3.5.输出文件处理

文件读取的逻辑非常简单：文件存在打开文件并写入数据，当文件不存在抛出异常。但是写入文件明显不能这么简单粗暴。新建一个JobInstance时最直观的操作是：存在同名文件就抛出异常，不存在则创建文件并写入数据。但是这样做显然有很大的问题，当批处理过程中出现问题需要restart，此时并不会从头开始处理所有的数据，而是要求文件存在并接着继续写入。为了确保这个过程FlatFileItemWriter默认会在新JobInstance运行时删除已有文件，而运行重启时继续在文件末尾写入。FlatFileItemWriter可以使用shouldDeleteIfExists、appendAllowed、shouldDeleteIfEmpty来有针对性的控制文件。

3.6.文件写入可执源码

文件写入主要代码在org.chenkui.spring.batch.sample.items.FlatFileWriter：

public class FlatFileWriter {

    private boolean useBuilder = true;

    @Bean
    public ItemWriter<MaxTemperatureEntiry> flatFileWriter() {
        BeanWrapperFieldExtractor<MaxTemperatureEntiry> fieldExtractor = new BeanWrapperFieldExtractor<>();
        fieldExtractor.setNames(new String[] { "siteId", "date", "temperature" }); //设置映射field
        fieldExtractor.afterPropertiesSet(); //参数检查

        DelimitedLineAggregator<MaxTemperatureEntiry> lineAggregator = new DelimitedLineAggregator<>();
        lineAggregator.setDelimiter(","); //设置输出分隔符
        lineAggregator.setFieldExtractor(fieldExtractor); //设置FieldExtractor处理器

        FlatFileItemWriter<MaxTemperatureEntiry> fileWriter = new FlatFileItemWriter<>();
        fileWriter.setLineAggregator(lineAggregator);
        fileWriter.setResource(new FileSystemResource("src/main/resources/out-data.csv")); //设置输出文件位置
        fileWriter.setName("outpufData");

        if (useBuilder) {//使用builder方式创建
            fileWriter = new FlatFileItemWriterBuilder<MaxTemperatureEntiry>().name("outpufData")
                .resource(new FileSystemResource("src/main/resources/out-data.csv")).lineAggregator(lineAggregator)
                .build();
        }
        return fileWriter;
    }
}

文件的写入过程与读取过程完全对称相反：先用FieldExtractor将对象转换为一个collection结构（列表），然后用lineAggregator将collection转化为带分隔符的字符串。

4.文件说明

代码中的测试数据来自数据分析交流项目bi-process-example，是NOAA的2015年全球天气监控数据。为了便于源码存储进行了大量的删减，原始数据有百万条，如有需要使用下列方式下载： curl -O ftp://ftp.ncdc.noaa.gov/pub/data/ghcn/daily/by_year/2015.csv.gz #数据文件 curl -O ftp://ftp.ncdc.noaa.gov/pub/data/ghcn/daily/ghcnd-stations.txt # 文件结构及类型说明
代码实现了读取文件、处理文件、写入文件的整个过程。处理文件的过程是只获取监控的最高温度信息（Type=TMAX），其他都过滤。
本案例的代码使用org.chenkui.spring.batch.sample.flatfile.FlatFileItemApplication::main方法运行，使用的是Command Runner的方式执行（运行方式的说明见Item概念及使用代码的命令行方式运行、Java内嵌运行）。