Spring Batch 不同数据库的多个阅读器

Question

我有一个现有的 spring 批处理项目，它在启动期间根据功能切换决策从 MySQL 或 ArangoDB（NoSql 数据库）读取数据，并执行一些过程并再次写回 MySQL/ArangoDB。

现在 MySQL 的阅读器配置如下所示，

@Bean
@Primary
@StepScope
public HibernatePagingItemReader reader(
        @Value("#{jobParameters[oldMetadataDefinitionId]}") Long oldMetadataDefinitionId) {

    Map<String, Object> queryParameters = new HashMap<>();
    queryParameters.put(Constants.OLD_METADATA_DEFINITION_ID, oldMetadataDefinitionId);


    HibernatePagingItemReader<Long> reader = new HibernatePagingItemReader<>();

    reader.setUseStatelessSession(false);
    reader.setPageSize(250);
    reader.setParameterValues(queryParameters);

    reader.setSessionFactory(((HibernateEntityManagerFactory) entityManagerFactory.getObject()).getSessionFactory());
    return reader;
}

我还有另一个像下面这样的 arango 阅读器，

@Bean
@StepScope
public ListItemReader arangoReader(
        @Value("#{jobParameters[oldMetadataDefinitionId]}") Long oldMetadataDefinitionId) {

    List<InstanceDTO> instanceList = new ArrayList<InstanceDTO>();

    PersistenceService arangoPersistence = arangoConfiguration
            .getPersistenceService());

    List<Long> instanceIds = arangoPersistence.getDefinitionInstanceIds(oldMetadataDefinitionId);

    instanceIds.forEach((instanceId) ->
    {
        InstanceDTO instanceDto = new InstanceDTO();
        instanceDto.setDefinitionID(oldMetadataDefinitionId);
        instanceDto.setInstanceID(instanceId);
        instanceList.add(instanceDto);

    });

    return new ListItemReader(instanceList);

}

我的步骤配置如下，

@Bean
@SuppressWarnings("unchecked")
public Step InstanceMergeStep(ListItemReader arangoReader, ItemWriter<MetadataInstanceDTO> arangoWriter,
        ItemReader<Long> mysqlReader, ItemWriter<Long> mysqlWriter) {

    Step step = null;
    if (arangoUsage) {
        step = steps.get("arangoInstanceMergeStep")


                .<Long, Long>chunk(1)

                .reader(arangoReader)

                .writer(arangoWriter)


                .faultTolerant()


                .skip(Exception.class)


                .skipLimit(10)

                .taskExecutor(stepTaskExecutor())

                .build();


        ((TaskletStep) step).registerChunkListener(chunkListener);
    }
    else {
        step = steps.get("mysqlInstanceMergeStep")


                .<Long, Long>chunk(1)

                .reader(mysqlReader)

                .writer(mysqlWriter)

                .faultTolerant()

                .skip(Exception.class)


                .skipLimit(failedSkipLimit)

                .taskExecutor(stepTaskExecutor())

                .build();


        ((TaskletStep) step).registerChunkListener(chunkListener);

    }

    return step;
}

MySQL 阅读器通过 HibernatePagingItemReader 支持分页，因此它可以处理数百万个项目而不会出现任何内存问题。

我想为 arango 阅读器实现相同的分页支持，每次迭代仅获取 250 个文档如何修改 arango 阅读器代码来实现这一点？

Answer 1

首先ListItemReader 的文档说 - 对测试有用 所以不要将它用于生产。返回一个ItemReader，而不是从你所有的阅读器bean，而不是实际的具体类型。

话虽如此，Spring Batch API 或 Spring Data 似乎不支持 Arango DB 。我能找到的最接近的是this

（我之前没有与 Arango DB 合作过）。

所以在我看来，你必须编写自己的自定义 arango 阅读器，通过可能实现抽象类来实现分页 - org.springframework.batch.item.database.AbstractPagingItemReader

如果不能通过扩展上述类来实现，您可能必须从头开始实现所有内容。 Spring Batch API 中的所有分页阅读器都扩展了这个抽象类，包括HibernatePagingItemReader。

另外，记住 arango 记录集应该有某种排序来实现分页，这样我们就可以区分 page - 0 和 page -1 等（类似于ORDER BY 子句，BETWEEN 运算符 & 小于，大于SQL 中的运算符等。还需要 FETCH FIRST XXX ROWS 或 LIMIT 子句之类的东西）。

自己实现并不是一项艰巨的任务，因为您必须计算可能的项目总数，对它们进行排序，然后分成几页，一次只获取一页。

查看 API 以获取诸如 - HibernatePagingItemReader 等实现的想法。

希望对你有帮助！！