Spring boot repository не сохраняет в БД при вызове из запланированного задания - PullRequest
Новая
       8

Spring boot repository не сохраняет в БД при вызове из запланированного задания

0 голосов
/ 31 января 2019

У меня есть приложение весенней загрузки, в котором мне нужно запланировать задание для чтения файлов из определенного каталога и сохранения данных в БД.

Я использовал Spring batch для обработки части файлов, так как количество файлов очень велико.

В приложении есть компонент с именем PraserStarer, который имеет метод с именем startParsing.Этот метод снабжен аннотацией @scheduled. 

@scheduled(fixedDelay = 60 * 1000)
public startParsing(){
    // start spring batch job
}

У меня есть интерфейс хранилища NewsRepositry, введенный в устройство записи первого шага пружинной партии.

Приложение имеет простой контроллер для ручного вызова метода startParsing.При вызове метода startParsing из контроллера все работает нормально.Пружинное пакетное задание запускается нормально, считывает файлы, записывает данные в БД и архивирует файлы.

Когда метод startParsing вызывается из среды планирования, весеннее пакетное задание запускается нормально, ичитать файлы, но ничего не хранится в БД.

Я подозреваю, что проблема здесь в том, что существует два разных контекста, один для части планирования, а другой для остальной части приложения.

По какой-то причине в контексте планирования нет диспетчера транзакций, который не приводит к переходу в БД.

1- Верны ли мои подозрения?

2- Если да, как я могу заставить менеджер транзакций загружаться в другой контекст?

РЕДАКТИРОВАТЬ

Код для начального класса анализатора ниже 

@Component
public class ParserStarter {

    @Autowired
    JobLauncher jobLauncher;

    @Value("${app.data_directory}")
    private String dataDir;

    @Autowired
    private ParserJobListener jobListener;

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    public Resource[] getResources() throws IOException {
        // return array of file resource to be processed
    }

//  @Scheduled(fixedDelay = 60 * 1000)
    public void startParsing() throws Exception {
        String jobName = System.currentTimeMillis() + " New Parser Job";

        JobParameters jobParameters = new JobParametersBuilder().addString("source", jobName).toJobParameters();

        jobLauncher.run(getParsingJob(), jobParameters);
    }

    @Bean(name="getParsingJob")
    private Job getParsingJob() throws IOException {

        jobListener.setResources(getResources());

        Step processingStep = jobListener.processingStep();

        Step archivingStep = jobListener.archivingStep();

        Job job = jobBuilderFactory.get("Store News").incrementer(new RunIdIncrementer())
                .listener(jobListener).start(processingStep).next(archivingStep).build();

        return job;
    }
}

Код для прослушивателя заданий ниже 

@Component
public class ParserJobListener extends JobExecutionListenerSupport {

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    private Resource[] resources;

    @Value("${app.archive_directory}")
    private String archiveDirectory;

    @Autowired
    private Writer writer;

    public MultiResourceItemReader<DataRecord> multiResourceItemReader() {
        MultiResourceItemReader<DataRecord> resourceItemReader = new MultiResourceItemReader<DataRecord>();
        resourceItemReader.setResources(resources);
        resourceItemReader.setDelegate(new Reader());
        return resourceItemReader;
    }

    public Step archivingStep() {
        FileArchivingTask archivingTask = new FileArchivingTask(resources, archiveDirectory);
        return stepBuilderFactory.get("Archiving step").tasklet(archivingTask).build();
    }

    public Step processingStep() {
        return stepBuilderFactory.get("Process news file").<DataRecord, DataRecord>chunk(1000)
                .reader(multiResourceItemReader()).writer(writer).build();
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
            System.out.println("Job finished")
        }
    }

    public void setResources(Resource[] resources) {
        this.resources = resources;
    }

}

Что такоеостается писатель, и он ниже 

@Component
public class Writer implements ItemWriter<DataRecord>{

    @Autowired
    private DataRepository dataRepo;

    @Override
    public void write(List<? extends DataRecord> items) throws Exception {
        dataRepo.saveAll(items);
    }

}

Edit 2

Я изменил метод записи писателя для индивидуального сохранения и очистки каждого элемента, как показано ниже 

@Transactional
    public void write(List<? extends GdeltRecord> items) throws Exception {
        for (GdeltRecord gdeltRecord : items) {
            dataRepo.saveAndFlush(gdeltRecord);
        }
//      dataRepo.saveAll(items);
 }

На этот раз приложение выдает исключение TransactionRequiredException: no transaction is in progress.

Вот полная трассировка стека исключения 

Caused by: javax.persistence.TransactionRequiredException: no transaction is in progress
    at org.hibernate.internal.SessionImpl.checkTransactionNeeded(SessionImpl.java:3552) ~[hibernate-core-5.3.7.Final.jar:5.3.7.Final]
    at org.hibernate.internal.SessionImpl.doFlush(SessionImpl.java:1444) ~[hibernate-core-5.3.7.Final.jar:5.3.7.Final]
    at org.hibernate.internal.SessionImpl.flush(SessionImpl.java:1440) ~[hibernate-core-5.3.7.Final.jar:5.3.7.Final]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_191]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_191]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_191]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_191]
    at org.springframework.orm.jpa.ExtendedEntityManagerCreator$ExtendedEntityManagerInvocationHandler.invoke(ExtendedEntityManagerCreator.java:350) ~[spring-orm-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at com.sun.proxy.$Proxy87.flush(Unknown Source) ~[na:na]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_191]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_191]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_191]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_191]
    at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:308) ~[spring-orm-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at com.sun.proxy.$Proxy87.flush(Unknown Source) ~[na:na]
    at org.springframework.data.jpa.repository.support.SimpleJpaRepository.flush(SimpleJpaRepository.java:533) ~[spring-data-jpa-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.data.jpa.repository.support.SimpleJpaRepository.saveAndFlush(SimpleJpaRepository.java:504) ~[spring-data-jpa-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_191]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_191]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_191]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_191]
    at org.springframework.data.repository.core.support.RepositoryComposition$RepositoryFragments.invoke(RepositoryComposition.java:359) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.data.repository.core.support.RepositoryComposition.invoke(RepositoryComposition.java:200) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.data.repository.core.support.RepositoryFactorySupport$ImplementationMethodExecutionInterceptor.invoke(RepositoryFactorySupport.java:644) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) [spring-aop-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.doInvoke(RepositoryFactorySupport.java:608) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.lambda$invoke$3(RepositoryFactorySupport.java:595) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.data.repository.core.support.RepositoryFactorySupport$QueryExecutorMethodInterceptor.invoke(RepositoryFactorySupport.java:595) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) [spring-aop-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at org.springframework.data.projection.DefaultMethodInvokingMethodInterceptor.invoke(DefaultMethodInvokingMethodInterceptor.java:59) ~[spring-data-commons-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) [spring-aop-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:294) ~[spring-tx-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:98) ~[spring-tx-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) [spring-aop-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139) ~[spring-tx-5.1.4.RELEASE.jar:5.1.4.RELEASE]
    ... 66 common frames omitted

Ответы [ 2 ]

0 голосов
/ 03 февраля 2019

Я попробовал подход, описанный в этом вопросе ( JpaItemWriter: транзакция не выполняется ), и он сработал для меня.

Я определил бин JpaTransactionManager и использовал его с конфигурацией шага. 

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

и в конфигурации шага 

    @Autowired
    JpaTransactionManager trxm;

    public Step processingStep(Resource[] resources) throws IOException {
        return stepBuilderFactory.get("Process CSV File")
                .transactionManager(trxm)
                .<DataRecord, DataRecord>chunk(1000)
                .reader(multiResourceItemReader()).writer(writer).build();
    }
0 голосов
/ 31 января 2019

Трудно проанализировать, что именно происходит, не видя полного кода.Однако, основываясь на документации , Spring использует чанк-ориентированную обработку.Вот что он говорит:

Как только количество прочитанных элементов становится равным интервалу фиксации, весь блок записывается через ItemWriter, а затем транзакция фиксируется.

Это может быть причиной того, что вы не видите никаких записей в БД сразу.

Что касается менеджера транзакций, вы можете определить его следующим образом (объяснено здесь ): 

@Bean
public Job sampleJob(JobRepository jobRepository, Step sampleStep) {
    return this.jobBuilderFactory.get("sampleJob")
                            .repository(jobRepository)
                .start(sampleStep)
                .build();
}

/**
 * Note the TransactionManager is typically autowired in and not needed to be explicitly
 * configured
 */
@Bean
public Step sampleStep(PlatformTransactionManager transactionManager) {
        return this.stepBuilderFactory.get("sampleStep")
                                .transactionManager(transactionManager)
                                .<String, String>chunk(10)
                                .reader(itemReader())
                                .writer(itemWriter())
                                .build();
}
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Нет похожих вопросов

...