Spring Batch中分段读取EDI文件转POJO的实现方案咨询

方案一：自定义流式EDI ItemReader + 轻量EDI解析库 #

核心思路是使用支持流式解析的EDI库，自定义Spring Batch的ItemStreamReader，逐段读取EDI内容并组装成业务POJO，实现分块处理，保留Spring Batch的批处理特性。

步骤1：引入流式EDI解析库 #

推荐使用ph-edi（轻量、支持流式的EDI解析库），添加Maven依赖：

<dependency>
    <groupId>com.helger</groupId>
    <artifactId>ph-edi</artifactId>
    <version>6.2.4</version>
</dependency>

步骤2：实现流式ItemReader #

实现ItemStreamReader接口，流式读取EDI段，按业务规则组装POJO，同时支持Spring Batch的重启功能：

public class EDIStreamingItemReader implements ItemStreamReader<Order> {
    private EDIReader ediReader;
    private File ediFile;
    private Order currentOrder;

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        try {
            ediReader = EDIReaderFactory.createEDIFACTReader(new FileInputStream(ediFile));
            // 恢复上次读取位置，支持任务重启
            if (executionContext.containsKey("edi.read.position")) {
                ediReader.setPosition(executionContext.getLong("edi.read.position"));
            }
        } catch (IOException e) {
            throw new ItemStreamException("无法打开EDI文件", e);
        }
    }

    @Override
    public Order read() throws Exception {
        while (ediReader.hasNext()) {
            EDIElement element = ediReader.next();
            if (element instanceof EDISegment) {
                EDISegment segment = (EDISegment) element;
                String segmentTag = segment.getSegmentTag();
                
                switch (segmentTag) {
                    case "UNH":
                        currentOrder = new Order();
                        currentOrder.setMessageId(segment.getField(0).getValue());
                        break;
                    case "LIN":
                        OrderLine line = new OrderLine();
                        line.setItemNumber(segment.getField(0).getValue());
                        currentOrder.addOrderLine(line);
                        break;
                    case "UNT":
                        // 完成一个订单组装，返回该POJO
                        Order completedOrder = currentOrder;
                        currentOrder = null;
                        return completedOrder;
                    case "BGM":
                        currentOrder.setOrderNumber(segment.getField(1).getValue());
                        break;
                    // 处理其他业务段，比如买家信息、日期等
                    default:
                        handleAdditionalSegments(segment, currentOrder);
                        break;
                }
            }
        }
        // 文件读取完毕，返回null
        return null;
    }

    private void handleAdditionalSegments(EDISegment segment, Order order) {
        // 示例：处理DTM段获取订单日期
        if ("DTM".equals(segment.getSegmentTag())) {
            order.setOrderDate(segment.getField(1).getValue());
        }
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        // 保存当前读取位置，支持任务重启
        executionContext.putLong("edi.read.position", ediReader.getPosition());
    }

    @Override
    public void close() throws ItemStreamException {
        try {
            if (ediReader != null) {
                ediReader.close();
            }
        } catch (IOException e) {
            throw new ItemStreamException("无法关闭EDI读取器", e);
        }
    }

    // 注入EDI文件路径
    public void setEdiFile(File ediFile) {
        this.ediFile = ediFile;
    }
}

步骤3：配置Spring Batch Job #

将自定义Reader与Processor、Writer结合，配置分块处理逻辑：

@Configuration
public class EDIJobConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public ItemReader<Order> ediItemReader() {
        EDIStreamingItemReader reader = new EDIStreamingItemReader();
        reader.setEdiFile(new File("/path/to/your/edi/file.edi"));
        return reader;
    }

    @Bean
    public ItemProcessor<Order, Order> ediItemProcessor() {
        return order -> {
            // 业务处理：验证字段、转换格式等
            order.setStatus("PROCESSING");
            return order;
        };
    }

    @Bean
    public ItemWriter<Order> ediItemWriter() {
        return orders -> {
            // 持久化操作：写入数据库、消息队列等
            orders.forEach(order -> System.out.println("处理订单：" + order.getOrderNumber()));
        };
    }

    @Bean
    public Step ediProcessingStep() {
        return stepBuilderFactory.get("ediProcessingStep")
                .<Order, Order>chunk(10) // 每10个订单为一个处理块
                .reader(ediItemReader())
                .processor(ediItemProcessor())
                .writer(ediItemWriter())
                .build();
    }

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

方案二：Smooks流式配置 + Spring Batch整合 #

如果坚持使用Smooks，可以通过启用流式解析模式避免一次性加载整个文件，结合Spring Batch实现分块处理：

步骤1：配置Smooks流式EDI映射 #

创建smooks-edi-config.xml，启用流式解析并定义EDI到POJO的映射：

<smooks-resource-list xmlns="http://www.milyn.org/xsd/smooks-1.1.xsd"
                      xmlns:edi="http://www.milyn.org/xsd/smooks/edi-1.4.xsd"
                      xmlns:jb="http://www.milyn.org/xsd/smooks/javabean-1.4.xsd">

    <!-- 启用流式EDI解析 -->
    <edi:reader mappingModel="classpath:edi-order-mapping.xml" stream="true"/>

    <!-- 映射EDI段到Order POJO -->
    <jb:bean beanId="order" class="com.example.Order" createOnElement="/order">
        <jb:value property="orderNumber" data="/order/orderNumber"/>
        <jb:wiring property="orderLines" beanId="orderLine"/>
    </jb:bean>

    <!-- 映射订单行到OrderLine POJO -->
    <jb:bean beanId="orderLine" class="com.example.OrderLine" createOnElement="/order/line">
        <jb:value property="itemNumber" data="/order/line/itemNumber"/>
    </jb:bean>
</smooks-resource-list>

步骤2：实现Smooks流式ItemReader #

public class SmooksEDIItemReader implements ItemStreamReader<Order> {
    private Smooks smooks;
    private Iterator<?> orderIterator;
    private File ediFile;

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        try {
            smooks = new Smooks("classpath:smooks-edi-config.xml");
            StreamSource ediSource = new StreamSource(new FileInputStream(ediFile));
            FilterResult result = new FilterResult();
            smooks.filterSource(ediSource, result);
            // 获取流式POJO迭代器
            orderIterator = result.getBeanContext().getBeanIterator("order");
        } catch (IOException | SmooksException e) {
            throw new ItemStreamException("初始化Smooks失败", e);
        }
    }

    @Override
    public Order read() throws Exception {
        if (orderIterator.hasNext()) {
            return (Order) orderIterator.next();
        }
        return null;
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        // 可选：添加重启位置记录逻辑
    }

    @Override
    public void close() throws ItemStreamException {
        smooks.close();
    }

    public void setEdiFile(File ediFile) {
        this.ediFile = ediFile;
    }
}

步骤3：配置Spring Batch Job #

与方案一类似，将该Reader配置到Job中并设置分块大小即可。

关键注意事项 #

• 流式解析：确保使用的EDI库支持流式读取，避免大文件一次性加载导致内存溢出。
• 边界处理：明确EDI段与业务POJO的映射边界（比如EDIFACT中UNH到UNT为一个完整业务消息），保证组装的POJO数据完整。
• 重启支持：实现ItemStreamReader的open和update方法，记录读取位置，支持Spring Batch的任务重启特性。

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