Spring批处理jpaPagingItemReader为什么有些行没有被读取？

Question

我正在使用Spring Batch(3.0.1.RELEASE)/ JPA和HSQLBD服务器数据库.我需要浏览整个表(使用分页)和更新项目(逐个).所以我使用了jpaPagingItemReader.但是当我运行该作业时,我可以看到跳过了一些行,并且跳过的行数等于页面大小.因为,如果我的表有12行而jpaPagingItemReader.pagesize = 3,则作业将显示:第1,2,3行,然后第7,8,9行(所以跳过第4,5,6行)...你能告诉我是什么我的代码/配置有问题,或者这可能是HSQLDB分页的问题？以下是我的代码:

[编辑]:问题在于我的ItemProcessor执行对POJO实体的修改.由于JPAPagingItemReader在每次读取之间进行刷新,因此实体会更新((这就是我想要的).但似乎游标分页也会增加(如日志中所示:行ID 4,5和6已经过跳过).我该如何处理这个问题？

@Configuration
@EnableBatchProcessing(modular=true)
public class AppBatchConfig {
  @Inject
  private InfrastructureConfiguration infrastructureConfiguration;  
  @Inject private JobBuilderFactory jobs;
  @Inject private StepBuilderFactory steps;

  @Bean  public Job job() {
     return jobs.get("Myjob1").start(step1()).build();
  }
  @Bean  public Step step1() {  
      return steps.get("step1")
                .<SNUserPerCampaign, SNUserPerCampaign> chunk(0)
                .reader(reader()).processor(processor()).build();   
  }
  @Bean(destroyMethod = "")
@JobScope 
public ItemStreamReader<SNUserPerCampaign> reader() String trigramme) {
    JpaPagingItemReader reader = new JpaPagingItemReader();
    reader.setEntityManagerFactory(infrastructureConfiguration.getEntityManagerFactory());
    reader.setQueryString("select t from SNUserPerCampaign t where t.isactive=true");
    reader.setPageSize(3));
    return reader;
}
 @Bean @JobScope
 public ItemProcessor<SNUserPerCampaign, SNUserPerCampaign> processor() {   
     return new MyItemProcessor();
 }
}

@Configuration
@EnableBatchProcessing
public class StandaloneInfrastructureConfiguration implements InfrastructureConfiguration {
 @Inject private EntityManagerFactory emf;  
 @Override
public EntityManagerFactory getEntityManagerFactory() {
    return emf;
}
}  

来自我的ItemProcessor:

@Override
public SNUserPerCampaign process(SNUserPerCampaign item) throws Exception {
    //do some stuff …
   //then if (condition) update the Entity pojo :   
   item.setModificationDate(new Timestamp(System.currentTimeMillis());
   item.setIsactive = false;

}

来自Spring xml配置文件:

<tx:annotation-driven transaction-manager="transactionManager" />     
<bean id="transactionManager" class="org.springframework.orm.jpa.JpaTransactionManager">
    <property name="entityManagerFactory" ref="entityManagerFactory" />
</bean>

<bean id="entityManagerFactory" class="org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean">
    <property name="dataSource" ref="dataSource" />
</bean>

<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
    <property name="driverClassName" value="org.hsqldb.jdbcDriver" />
    <property name="url" value="jdbc:hsqldb:hsql://localhost:9001/MYAppDB" />
    <property name="username" value="sa" />
    <property name="password" value="" />
</bean>

跟踪/日志汇总:

11:16:05.728 TRACE MyItemProcessor - item processed: snUserInternalId=1]
11:16:06.038 TRACE MyItemProcessor - item processed: snUserInternalId=2]
11:16:06.350 TRACE MyItemProcessor - item processed: snUserInternalId=3]

11:16:06.674 DEBUG SQL- update SNUSER_CAMPAIGN  set ...etc...
11:16:06.677 DEBUG SQL- update SNUSER_CAMPAIGN  set ...etc...
11:16:06.679 DEBUG SQL- update SNUSER_CAMPAIGN  set ...etc...

11:16:06.681 DEBUG SQL- select ...etc... from  SNUSER_CAMPAIGN snuserperc0_ 

11:16:06.687 TRACE MyItemProcessor - item processed: snUserInternalId=7]
11:16:06.998 TRACE MyItemProcessor - item processed: snUserInternalId=8]
11:16:07.314 TRACE MyItemProcessor - item processed: snUserInternalId=9]

Answer 1

org.springframework.batch.item.database.JpaPagingItemReader创建自己的entityManager实例

(来自org.springframework.batch.item.database.JpaPagingItemReader #doOpen):

entityManager = entityManagerFactory.createEntityManager(jpaPropertyMap);

如果你在一个事务中,就像它似乎一样,读者实体不是分离的(来自org.springframework.batch.item.database.JpaPagingItemReader #doReadPage):

    if (!transacted) {
        List<T> queryResult = query.getResultList();
        for (T entity : queryResult) {
            entityManager.detach(entity);
            results.add(entity);
        }//end if
    } else {
        results.addAll(query.getResultList());
        tx.commit();
    }

因此,当您将项目更新为处理器或编写器时,此项目仍由读者的entityManager管理.

当项目阅读器读取下一个数据块时,它会将上下文刷新到数据库.

所以,如果我们看一下你的情况,在第一块数据处理之后,我们就在数据库中:

|id|active
|1 | false
|2 | false
|3 | false

org.springframework.batch.item.database.JpaPagingItemReader使用limit和offset来检索分页数据.所以读者创建的下一个选择如下:

select * from table where active = true offset 3 limits 3. 

Reader将错过ID为4,5,6的项目,因为它们现在是数据库检索的第一行.

作为一种解决方法,您可以使用jdbc实现(org.springframework.batch.item.database.JdbcPagingItemReader),因为它不使用limit和offset.它基于有序列(通常是id列),因此您不会遗漏任何数据.当然,您必须将数据更新到编写器中(使用JPA纯JDBC实现)

读者会更加冗长:

@Bean
public ItemReader<? extends Entity> reader() {
    JdbcPagingItemReader<Entity> reader = new JdbcPagingItemReader<Entity>();
    final SqlPagingQueryProviderFactoryBean sqlPagingQueryProviderFactoryBean = new SqlPagingQueryProviderFactoryBean();
    sqlPagingQueryProviderFactoryBean.setDataSource(dataSource);
    sqlPagingQueryProviderFactoryBean.setSelectClause("select *");
    sqlPagingQueryProviderFactoryBean.setFromClause("from <your table name>");
    sqlPagingQueryProviderFactoryBean.setWhereClause("where active = true");
    sqlPagingQueryProviderFactoryBean.setSortKey("id");
    try {
        reader.setQueryProvider(sqlPagingQueryProviderFactoryBean.getObject());
    } catch (Exception e) {
        e.printStackTrace();
    }
    reader.setDataSource(dataSource);
    reader.setPageSize(3);
    reader.setRowMapper(new BeanPropertyRowMapper<Entity>(Entity.class));
    return reader;

Answer 2

我遇到了同样的情况，我的阅读器是一个 JpaPagingItemReader，它查询写入器中更新的字段。因此，由于页面窗口正在进行，而已读取的项目不再处于阅读器范围内，因此跳过了一半需要更新的项目。

对我来说最简单的解决方法是重写 JpaPagingItemReader 上的 getPage 方法以始终返回第一页。

JpaPagingItemReader<XXXXX> jpaPagingItemReader = new JpaPagingItemReader() {
    @Override
    public int getPage() {
        return 0;
    }
};

Answer 3

需要注意的几点：

1. 从 返回的所有实体JpaPagingItemReader都是分离的。我们通过以下两种方式之一来实现这一点。我们要么在查询页面之前创建一个事务，然后提交该事务（分离与该EntityManager事务关联的所有实体），要么我们显式调用entityManager.detach. 我们这样做是为了可以正确执行重试和跳过等功能。
2. 虽然您没有在处理器中发布所有代码，但我的预感是在该//do some stuff部分中，您的项目正在重新附加，这就是更新发生的原因。但是，由于无法看到该代码，我无法确定。
3. 在任何一种情况下，ItemWriter都应该使用显式。事实上，我认为这是一个ItemWriter在使用 java 配置时我们不需要的错误（我们为 XML 做）。
4. 对于丢失记录的特定问题，您需要记住任何*PagingItemReaders都不使用游标。它们都对每一页数据执行独立的查询。因此，如果您更新每个页面之间的基础数据，它可能会对未来页面中返回的项目产生影响。例如，如果我的分页查询指定where val1 > 4并且我有一条记录，即 val1 从 1 变为 5，则在块 2 中，该项目可能会被返回，因为它现在符合条件。如果您需要更新 where 子句中的值（从而影响您要处理的数据集中的内容），最好添加某种可以查询的已处理标志。