Spring Batch实现单条数据分存Work表与Identifier键值表

嘿，这个场景我之前做过类似的，刚好你已经在用CompositeItemWriter，那咱们只需要加一步拆分处理，再给CompositeItemWriter配两个子Writer就行，我给你一步步讲清楚怎么实现：

核心思路 #

咱们的目标是把单个MetsModsDef对象拆成两类数据：一条Work记录 + N条Identifier记录。所以流程应该是：

1. 用自定义ItemProcessor把MetsModsDef转换成一个包含Work和Identifier列表的复合对象
2. 让CompositeItemWriter里的两个子Writer分别处理Work数据和Identifier列表数据

具体实现步骤 #

1. 定义对应数据库表的实体类 #

先把Work和Identifier的实体类写好，和数据库字段一一对应：

// Work实体类，对应Work表
public class Work {
    private int id;
    private String title;
    private String path;
    
    // 省略getter、setter、全参/无参构造方法
}

// Identifier实体类，对应Identifier表
public class Identifier {
    private int workId; // 关联Work表的id
    private String identityType; // 对应Properties的key
    private String identityValue; // 对应Properties的value
    
    // 省略getter、setter、全参/无参构造方法
}

2. 编写拆分用的ItemProcessor #

这个处理器的作用是把MetsModsDef拆成Work和Identifier列表，然后封装到一个自定义容器类里（这样后续Writer能方便拿到两类数据）：

// 自定义容器类，用来封装拆分后的Work和Identifier列表
public class WorkWithIdentifiers {
    private Work work;
    private List<Identifier> identifiers;
    
    // 省略getter、setter、构造方法
}

// 拆分处理器实现
public class MetsModsSplitterProcessor implements ItemProcessor<MetsModsDef, WorkWithIdentifiers> {
    @Override
    public WorkWithIdentifiers process(MetsModsDef item) throws Exception {
        // 1. 构建Work对象
        Work work = new Work();
        work.setId(item.getId());
        work.setTitle(item.getTitle());
        work.setPath(item.getPath());

        // 2. 遍历Properties，构建Identifier列表
        List<Identifier> identifiers = new ArrayList<>();
        Properties identifiersProps = item.getIdentifiers();
        for (String key : identifiersProps.stringPropertyNames()) {
            Identifier identifier = new Identifier();
            identifier.setWorkId(item.getId()); // 绑定关联的Work id
            identifier.setIdentityType(key);
            identifier.setIdentityValue(identifiersProps.getProperty(key));
            identifiers.add(identifier);
        }

        // 3. 封装成复合对象返回
        WorkWithIdentifiers result = new WorkWithIdentifiers();
        result.setWork(work);
        result.setIdentifiers(identifiers);
        return result;
    }
}

3. 编写两个子Writer分别处理Work和Identifier #

接下来写两个Writer，分别负责往Work表和Identifier表插数据：

// Work表的Writer
public class WorkItemWriter implements ItemWriter<WorkWithIdentifiers> {
    private final JdbcTemplate jdbcTemplate;

    // 构造注入JdbcTemplate，也可以用Spring Data JPA的Repository
    public WorkItemWriter(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public void write(List<? extends WorkWithIdentifiers> items) throws Exception {
        String insertSql = "INSERT INTO Work (id, title, path) VALUES (?, ?, ?)";
        // 批量插入，提升效率
        List<Object[]> batchParams = items.stream()
                .map(item -> new Object[]{
                        item.getWork().getId(),
                        item.getWork().getTitle(),
                        item.getWork().getPath()
                })
                .collect(Collectors.toList());
        jdbcTemplate.batchUpdate(insertSql, batchParams);
    }
}

// Identifier表的Writer
public class IdentifierItemWriter implements ItemWriter<WorkWithIdentifiers> {
    private final JdbcTemplate jdbcTemplate;

    public IdentifierItemWriter(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public void write(List<? extends WorkWithIdentifiers> items) throws Exception {
        String insertSql = "INSERT INTO Identifier (work, identitytype, identityValue) VALUES (?, ?, ?)";
        List<Object[]> batchParams = new ArrayList<>();
        
        // 遍历所有复合对象，把每个Identifier都转成批量参数
        for (WorkWithIdentifiers item : items) {
            for (Identifier identifier : item.getIdentifiers()) {
                batchParams.add(new Object[]{
                        identifier.getWorkId(),
                        identifier.getIdentityType(),
                        identifier.getIdentityValue()
                });
            }
        }
        
        jdbcTemplate.batchUpdate(insertSql, batchParams);
    }
}

4. 组装Spring Batch的Job和Step #

最后把这些组件都配置到Spring Batch的Job里，重点是把两个子Writer加到CompositeItemWriter中：

@Configuration
public class XmlProcessingBatchConfig {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private DataSource dataSource;

    // 你的XML ItemReader，这里假设已经配置好（比如用StaxEventItemReader读取XML）
    @Bean
    public ItemReader<MetsModsDef> xmlItemReader() {
        return new StaxEventItemReaderBuilder<MetsModsDef>()
                .resource(new ClassPathResource("input-data.xml"))
                .rootElementName("metsModsDef")
                .unmarshaller(new Jaxb2Marshaller() {{
                    setClassesToBeBound(MetsModsDef.class);
                }})
                .build();
    }

    // 注册拆分处理器
    @Bean
    public ItemProcessor<MetsModsDef, WorkWithIdentifiers> splitterProcessor() {
        return new MetsModsSplitterProcessor();
    }

    // 注册Work表Writer
    @Bean
    public ItemWriter<WorkWithIdentifiers> workItemWriter() {
        return new WorkItemWriter(new JdbcTemplate(dataSource));
    }

    // 注册Identifier表Writer
    @Bean
    public ItemWriter<WorkWithIdentifiers> identifierItemWriter() {
        return new IdentifierItemWriter(new JdbcTemplate(dataSource));
    }

    // 配置CompositeItemWriter，组合两个子Writer
    @Bean
    public CompositeItemWriter<WorkWithIdentifiers> compositeItemWriter() {
        CompositeItemWriter<WorkWithIdentifiers> compositeWriter = new CompositeItemWriter<>();
        compositeWriter.setDelegates(Arrays.asList(workItemWriter(), identifierItemWriter()));
        return compositeWriter;
    }

    // 配置Step
    @Bean
    public Step processXmlStep() {
        return stepBuilderFactory.get("processXmlStep")
                .<MetsModsDef, WorkWithIdentifiers>chunk(100) // 批次大小根据数据量和性能调整
                .reader(xmlItemReader())
                .processor(splitterProcessor())
                .writer(compositeItemWriter())
                .build();
    }

    // 配置Job
    @Bean
    public Job xmlToDbJob() {
        return jobBuilderFactory.get("xmlToDbJob")
                .start(processXmlStep())
                .build();
    }
}

几个关键注意点 #

• 事务一致性：Spring Batch默认每个Chunk一个事务，如果Work插入成功但Identifier插入失败，整个Chunk会回滚，保证数据的一致性，这符合大多数业务场景的需求。
• 性能优化：代码里用了jdbcTemplate.batchUpdate批量插入，比单条插入效率高很多，尤其是数据量较大时。
• 自增主键场景：如果Work的id是数据库自增的，那需要调整逻辑：插入Work后获取自增id，再设置到Identifier的workId上。可以用GeneratedKeyHolder来获取自增主键，然后在Processor里处理（或者在WorkWriter里处理后把id传递给IdentifierWriter，这时候可能需要调整容器类的设计）。

内容的提问来源于stack exchange，提问作者call me carrot