使用 JPA 存储库的春季批处理 ItemWriter 存在问题

Posted 2023-02-26

【中文标题】使用 JPA 存储库的春季批处理 ItemWriter 存在问题

【英文标题】：Persist issue with a spring batch ItemWriter using a JPA repository

【发布时间】：2016-11-12 05:47:08

【问题描述】：

我对依赖 JPA 存储库来更新数据的 Spring Batch ItemWriter 有疑问。

这里是：

@Component
public class MessagesDigestMailerItemWriter implements ItemWriter<UserAccount> 

    private static final Logger log = LoggerFactory.getLogger(MessagesDigestMailerItemWriter.class);

    @Autowired
    private MessageRepository messageRepository;

    @Autowired
    private MailerService mailerService;

    @Override
    public void write(List<? extends UserAccount> userAccounts) throws Exception 
        log.info("Mailing messages digests and updating messages notification statuses");

        for (UserAccount userAccount : userAccounts) 
            if (userAccount.isEmailNotification()) 
                mailerService.mailMessagesDigest(userAccount);
            
            for (Message message : userAccount.getReceivedMessages()) 
                message.setNotificationSent(true);
                messageRepository.save(message);//NOT SAVING!!
            
        
    

这是我的Step 配置：

@Configuration
public class MailStepConfiguration 

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private EntityManagerFactory entityManagerFactory;

    @Autowired
    private MessagesDigestMailerItemWriter itemWriter;

    @Bean
    public Step messagesDigestMailingStep() 
        return stepBuilderFactory.get("messagesDigestMailingStep")//
                .<UserAccount, UserAccount> chunk(1)//
                .reader(jpaPagingItemReader(entityManagerFactory))//
                .writer(itemWriter)//
                .build();
    

    @Bean(destroyMethod = "")
    @StepScope
    public static ItemReader<UserAccount> jpaPagingItemReader(EntityManagerFactory entityManagerFactory) 
        final JpaPagingItemReader<UserAccount> reader = new JpaPagingItemReader<>();
        reader.setEntityManagerFactory(entityManagerFactory);
        reader.setQueryString("SELECT ua FROM UserAccount ua JOIN FETCH ua.receivedMessages msg WHERE msg.notificationSent = false AND msg.messageRead = false");
        return reader;
    

为了完整起见，这是我的 Spring Boot 配置：

@Configuration
@EnableBatchProcessing
@EnableAutoConfiguration
@ComponentScan("com.bignibou.batch.configuration")
public class Batch 
    public static void main(String[] args) 
        System.exit(SpringApplication.exit(new SpringApplicationBuilder(Batch.class).web(false).run(args)));
    

和我的数据源配置：

@Configuration
@EnableJpaRepositories( "com.bignibou.repository" )
@EntityScan("com.bignibou.domain")
public class DatasourceConfiguration 

    @Bean
    @ConfigurationProperties("spring.datasource.batch")
    public DataSource batchDatasource() 
        return DataSourceBuilder.create().build();
    

    @Bean
    @Primary
    @ConfigurationProperties("spring.datasource.application")
    public DataSource applicationDatasource() 
        return DataSourceBuilder.create().build();
    

我注意到执行流程进入了 ItemWriter 的 write 方法，messageRepository.save(message); 确实被执行但数据没有更新。

我怀疑这是一个交易问题，但我不知道如何解决这个问题...

edit：我忘了说我有两个 Postgres 数据库：

一个用于作业存储库数据 另一个用于应用程序数据。

我可以确认数据已写入作业存储库数据库。问题在于应用程序数据。考虑到我有两个 PG 数据库这一事实，我需要使用分布式事务吗？

【问题讨论】：

似乎唯一的解决方案是使用分布式 tx 管理器。我在这里打开了另一个问题：***.com/questions/38323207

嗨@balteo，你有解决方案吗？我正面临着确切的问题。如果您能分享解决方案，那就太好了。

您好@balteo，您找到解决此问题的方法了吗？我正在尝试以与您类似的方式在 writer 中使用 Repository，但结果同样令人不满意...... :-(

【参考方案1】：

我在这里打开了一个问题：

https://jira.spring.io/browse/BATCH-2642

原则上，帮助我们的是像这样配置主事务管理器：

@Configuration
public class JpaConfig 

    private final DataSource dataSource;

    @Autowired
    public JpaConfig(@Qualifier("dataSource") DataSource dataSource) 
        this.dataSource = dataSource;
    

    @Bean
    @Primary
    public JpaTransactionManager jpaTransactionManager() 
        final JpaTransactionManager transactionManager = new JpaTransactionManager();
        transactionManager.setDataSource(dataSource);
        return transactionManager;
    

然后在配置步骤时使用事务管理器的自动装配实例，如下所示：

@Autowired
private PlatformTransactionManager transactionManager;

private TaskletStep buildTaskletStep() 
        return stepBuilderFactory.get("SendCampaignStep")
                    .<UserAccount, UserAccount>chunk(pushServiceConfiguration.getCampaignBatchSize())
                    .reader(userAccountItemReader)
                    .processor(userAccountItemProcessor)
                    .writer(userAccountItemWriter)
                    .transactionManager(transactionManager)
                    .build();
    

数据现在已正确持久化，但仍有一些我没有完全理解的魔力...

【讨论】：

实际上我正面临这个问题，我对这个 TransactionManager 和 Spring Data JPA 感到困惑。如果我只设置dataSource，可以吗？它不会影响 Spring Data JPA 默认配置吗？我是否需要设置一些其他的东西，比如EntityManagerFactory？

Thaaaaaaaaaaaaaaaaaaaaaa 太感谢你了！！！你救了我的命

【参考方案2】：

你应该在你的主课上@EnableTransactionManagement。我相信 Spring Boot 会为你创建事务管理器，但如果你想覆盖默认值，你可以want to configure it explicitly。

Spring Batch 提供APIs for changing transaction attributes。

【讨论】：

非常感谢您的回答。我试过@EnableTransactionManagement 无济于事。我相信它适用于 @Transaction，不建议在 Spring Batch 中使用。