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

【问题标题】：Spring batch jpaPagingItemReader why some rows are not read?Spring批处理jpaPagingItemReader为什么有些行没有被读取？
【发布时间】：2014-12-18 01:14:20
【问题描述】：

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

[EDIT]：问题出在我的 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]

【问题讨论】：

您如何确认正在读取哪些行？您可以提供任何堆栈跟踪/日志信息吗？
我意识到问题是我的 ItemProcessor 改变了实体 pojo，我通过添加跟踪来编辑我的问题
我有点困惑。从JpaPagingItemReader 返回的项目是分离的，因此在没有明确写入的情况下不应更新它们。能否提供完整的作业配置？
我明天会尝试添加更多细节。无论如何，我在 Spring JPAPagingItemreader 的源代码中可以看到，在阅读下一页时，这些项目似乎是分离的（通过对上下文进行刷新和清除）。这就是为什么我不需要 ItemWritter 来持久化我的更新。一个重要的细节可能是，在我的项目处理器中，我更新了我的 SELECT where 子句中的布尔值；不知道这样能不能修改分页光标。
冲洗和清除只是一开始的清理。在JpaPagingItemReader 结束时，我们循环遍历元素并显式分离它们，或者提交事务以便分离它们。无论哪种方式，从JpaPagingItemReader 返回的实体都应该处于分离状态。您可以发布完整工作的配置吗？

标签： jpa spring-batch

【解决方案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 使用限制和偏移来检索分页数据。所以读者创建的下一个选择看起来像：

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

阅读器将错过 id 为 4、5、6 的项目，因为它们现在是数据库检索到的第一行。

作为一种解决方法，您可以做的是使用 jdbc 实现 (org.springframework.batch.item.database.JdbcPagingItemReader)，因为它不使用限制和偏移。它基于已排序的列（通常是 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;

【讨论】：

听起来不错。我需要做更多的测试才能验证这个解决方案。因为我的最终 SQL 会更复杂（复合主键，where 子句包括“select in”......）我还必须从 JdbcPagingItemReader 继承来添加一些额外的操作。还是谢谢……
对于复合键，您可以使用 org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean#setSortKeys 。自从 Michael Minella 的修复（jira.spring.io/browse/BATCH-2079，thanx Michael！）以来，它就可以使用别名。
上面关于不分离的说法是错误的。 tx.commit() 应该将所有与事务关联的实体（在 doReadPage 方法的开头创建）分离。
我也预料到了，但是使用调试器，我可以看到阅读器的 entityManager 包含在处理器/编写器中更新的实体。在我的例子中，实体没有分离。
我同意似乎即使在 JpaPagingItemReader 提交之后，这些项目仍在 entityManager 中，并且在读取下一页时会更新，在我的日志文件中我没有看到一些可能对应的 SQL 请求到“重新附加”请求。

【解决方案2】：

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

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

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

【讨论】：

这适用于单线程方法，但与 taskExecutor 一起使用时可能会很危险。

【解决方案3】：

需要注意的几点：

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

【讨论】：

感谢您的帖子。在我的处理器中，我处理该项目并更改一些值（包括选择条件中的一些值）。所以这解释了这个问题，但没有看到任何可以“重新附加”项目的操作或日志跟踪，似乎它们在处理过程中仍在事务中，直到下一页正在阅读。
它太大了，但昨天即使我几乎把所有都放在评论中，我也可以重现这个问题。项目处理器正在使用 Pojo 实体的 getter 和 setter；问题的发生是因为，如前所述，我修改了 SELECT 请求条件中的一些字段。
我很困惑。使用状态列标记已处理的项目并在阅读器的 where 子句中使用它不是任何数据库阅读器的常见模式吗？这样做的首选方法是什么？如果 *PagingrItemReaders 对每个页面执行独立的查询，为什么还要跟踪页面呢？ @Yves-Marie L's hack to override getPage() 似乎有效 - 但似乎应该有更好的方法来处理这个问题。我错过了什么？

【解决方案4】：

我遇到了同样的问题，即根据 pageSize 跳过行。例如，如果我将 pageSize 设置为 2，它将读取 2、忽略 2、读取 2、忽略 2 等。

我正在构建一个守护程序处理器来轮询“请求”数据库表以获取处于“等待处理”状态的记录。该守护程序旨在永远在后台运行。

我有一个在@NamedQuery 中定义的“状态”字段，它会选择状态为“10”的记录：等待处理。处理记录后，状态字段将更新为“20”：错误或“30”：成功。事实证明这是问题的原因 - 我正在更新查询中定义的字段。如果我引入了一个 'processedField' 并更新了它而不是 'status' 字段，那么没问题 - 所有的记录都会被读取。

作为更新状态字段的可能解决方案，我将MaxItemCount设置为与PageSize相同；这在步骤完成之前正确更新了记录。然后我继续执行该步骤，直到发出停止守护程序的请求。好的，可能不是最有效的方法（但我仍然受益于 JPA 提供的易用性），但我认为使用 JdbcPagingItemReader 可能会更好（如上所述 - 谢谢！）。欢迎对这个批处理数据库轮询问题的最佳方法提出意见:)

【讨论】：