java - 在特定步骤下，单个处理器中的多个实体管理器(为多个数据库连接使用不同的数据源):Spring Batch

我试图在处理器的Spring Batch项目中使用2个实体管理器，一个用于读取，另一个用于将数据持久存储在数据库中。
代码如下

@StepScope
@ComponentScan("com.airtel.billreminder.model")
@Transactional
@Slf4j
public class MyItemProcessor implements ItemProcessor<VoltSiMaster,VoltSiMaster>{

    @PersistenceContext(unitName="voltuser")
    EntityManager emvoltUser;

    @PersistenceContext(unitName="voltreporting")
    EntityManager emvoltUserReporting;

    @Value("#{jobParameters['notificationNumber']}")
    public Long notificationNumber;

    @Value("#{jobParameters['dayBeforeNotificationValue']}")
    public Long dayBeforeNotificationValue;

    @Value("#{jobParameters['firstReminderBeforeDueDate']}")
    public Long firstReminderBeforeDueDate;

    private ReminderHistory reminderHistory;

    @Transactional
    @Override
    public VoltSiMaster process(VoltSiMaster voltSiMaster) throws Exception {
        log.info("Inside Processor");
        perpareNotifications(voltSiMaster,notificationNumber,dayBeforeNotificationValue);
        return voltSiMaster;
    }

    @Transactional
    private void perpareNotifications(VoltSiMaster siMaster, long notificationNumber,long dayBeforeNotificationValue) {
        try {
                /*business logic*/
                if(notificationNumber>0)
                {
                    txnDone = checkIsTxnAlreadyDone(siMaster);
                }
                if (!txnDone) {

                    String[] communincationTypesArray = communincationTypes.split(Pattern.quote("|"));
                    for (String communincationType : communincationTypesArray) {
                        log.info("Before sending notification");
                        if(siMaster.getAmount()>0)
                        sendNotification(notificationNumber,siMaster,
                                customerResponse, communincationType);
                        else
                        log.info("Amount is "+ siMaster.getBillerName() +" "+siMaster.getAmount()+"so not sending notification");
                    }
                }
                else
                {
                    log.info("User has Already Paid Bill so no need to send notification");
                }
        }
    }

    private boolean checkIsTxnAlreadyDone(VoltSiMaster siMaster) {
        String sqlString = "select * from "+Schemas.VOLT_USER_REPORTING+".volt_txn_log  where txn_date_time >= (?1-"+firstReminderBeforeDueDate+")  and CUSTOM_COL_3=?2 and uc_id=?3";
        try {
            Query txnCheckQuery = emvoltUserReporting.createNativeQuery(sqlString, VoltTxnLog.class);
            txnCheckQuery.setParameter(1, siMaster.getBillDueDate());
            txnCheckQuery.setParameter(2, siMaster.getReference1());
            txnCheckQuery.setParameter(3, siMaster.getUcId());
            log.info("Query to fire : {} " + txnCheckQuery.toString());
            @SuppressWarnings("unchecked")
            List<VoltTxnLog> resultList = txnCheckQuery.getResultList();
            if (resultList != null) {
                if (resultList.size() > 0)
                {
                    for(VoltTxnLog vtxnlog: resultList)
                    {
                        if(vtxnlog.getTxnStatus().equals("0") || vtxnlog.getTxnStatus().equals("2"))
                            return true;
                    }
                }
                else
                    return false;
            } else
                return false;
        } catch (Exception e) {
            log.info("Error: "+e.getMessage());
        }
        return false;
    }

    @Transactional
    private void sendNotification(long notificationNumber,VoltSiMaster siMaster,CusDetResponse customerResponse, String communincationType) {
        /*business logic*/
        emvoltUser.persist(reminderHistory);
        emvoltUser.flush();
    }

}

第一个实体管理器的配置文件：

import javax.persistence.EntityManagerFactory;
import javax.persistence.PersistenceContext;
import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.orm.jpa.JpaTransactionManager;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;

@Configuration
@EnableTransactionManagement
@EnableJpaRepositories(entityManagerFactoryRef = "voltEntityManagerFactory",
transactionManagerRef = "voltTransactionManager", basePackages = {"com.airtel.billreminder.model.voltUser"})
public class VoltUserDbConfig {

    @Primary
    @Bean(name = "voltDataSource")
    @ConfigurationProperties(prefix = "volt.datasource")
    public DataSource dataSource() {
        return DataSourceBuilder.create().build();
    }

    @Primary
    @PersistenceContext(unitName="voltuser")
    @Bean(name = "voltEntityManagerFactory")
    public LocalContainerEntityManagerFactoryBean barEntityManagerFactory(
            EntityManagerFactoryBuilder builder, @Qualifier("voltDataSource") DataSource dataSource) {
        return builder.dataSource(dataSource).packages("com.airtel.billreminder.model.voltUser").persistenceUnit("voltuser")
                .build();
    }

    @Primary
    @ConditionalOnMissingBean(name = "transactionManager")
    @ConditionalOnBean(DataSource.class)
    @Bean(name = "voltTransactionManager")
    public PlatformTransactionManager barTransactionManager(
            @Qualifier("voltEntityManagerFactory") EntityManagerFactory barEntityManagerFactory) {
        return new JpaTransactionManager(barEntityManagerFactory);
    }

}

第二个实体管理器的配置文件：

import javax.persistence.EntityManagerFactory;
import javax.persistence.PersistenceContext;
import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.orm.jpa.JpaTransactionManager;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;

@Configuration
@EnableTransactionManagement
@EnableJpaRepositories(entityManagerFactoryRef = "voltReportingEntityManagerFactory",
transactionManagerRef = "voltReportingTransactionManager", basePackages = {"com.airtel.billreminder.model.voltUserReporting"})
public class VoltUserReportingDbConfig {

    @Bean(name = "voltReportingDataSource")
    @ConfigurationProperties(prefix = "voltreporting.datasource")
    public DataSource dataSource() {
        return DataSourceBuilder.create().build();
    }

    @PersistenceContext(unitName="voltreporting")
    @Bean(name = "voltReportingEntityManagerFactory")
    public LocalContainerEntityManagerFactoryBean barEntityManagerFactory(
            EntityManagerFactoryBuilder builder, @Qualifier("voltReportingDataSource") DataSource dataSource) {
        return builder.dataSource(dataSource).packages("com.airtel.billreminder.model.voltUserReporting").persistenceUnit("voltreporting")
                .build();
    }

    @ConditionalOnMissingBean(name = "transactionManager")
    @ConditionalOnBean(DataSource.class)
    @Bean(name = "voltReportingTransactionManager")
    public PlatformTransactionManager barTransactionManager(
            @Qualifier("voltReportingEntityManagerFactory") EntityManagerFactory barEntityManagerFactory) {
        return new JpaTransactionManager(barEntityManagerFactory);
    }

}

Spring Batch初始化了自己的事务管理器，该事务管理器我使用@ConditionalOnMissingBean批注处理，因为它先前未创建我的自定义实体管理器

现在运行此代码使我在emvoltUser.flush（）处出现以下异常，即使我使用@Transactional时也应自行处理事务

javax.persistence.TransactionRequiredException: no transaction is in progress
at org.hibernate.internal.SessionImpl.checkTransactionNeeded(SessionImpl.java:3505)
at org.hibernate.internal.SessionImpl.doFlush(SessionImpl.java:1427)
at org.hibernate.internal.SessionImpl.flush(SessionImpl.java:1423)
at sun.reflect.GeneratedMethodAccessor56.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.orm.jpa.ExtendedEntityManagerCreator$ExtendedEntityManagerInvocationHandler.invoke(ExtendedEntityManagerCreator.java:350)
at com.sun.proxy.$Proxy109.flush(Unknown Source)
at sun.reflect.GeneratedMethodAccessor56.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:305)
at com.sun.proxy.$Proxy109.flush(Unknown Source)
at com.airtel.billreminder.notificationStep.MyItemProcessor.sendNotification(MyItemProcessor.java:233)
at com.airtel.billreminder.notificationStep.MyItemProcessor.perpareNotifications(MyItemProcessor.java:135)
at com.airtel.billreminder.notificationStep.MyItemProcessor.process(MyItemProcessor.java:112)
at com.airtel.billreminder.notificationStep.MyItemProcessor.process(MyItemProcessor.java:58)
at com.airtel.billreminder.notificationStep.MyItemProcessor$$FastClassBySpringCGLIB$$29bbf641.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:294)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:98)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
at com.airtel.billreminder.notificationStep.MyItemProcessor$$EnhancerBySpringCGLIB$$7ddd0f69.process(<generated>)
at com.airtel.billreminder.notificationStep.MyItemProcessor$$FastClassBySpringCGLIB$$29bbf641.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:746)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:136)
at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:124)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
at com.airtel.billreminder.notificationStep.MyItemProcessor$$EnhancerBySpringCGLIB$$ab91a70.process(<generated>)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.doProcess(SimpleChunkProcessor.java:126)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.transform(SimpleChunkProcessor.java:303)
at org.springframework.batch.core.step.item.SimpleChunkProcessor.process(SimpleChunkProcessor.java:202)
at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:75)
at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:406)
at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:330)
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:140)
at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:272)
at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:81)
at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:375)
at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:145)
at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257)
at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200)
at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:66)
at
org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:67)
at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:169)
at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:144)
at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:136)
at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:308)
at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:141)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:134)
at sun.reflect.GeneratedMethodAccessor66.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:343)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:197)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration$PassthruAdvice.invoke(SimpleBatchConfiguration.java:127)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
at com.sun.proxy.$Proxy112.run(Unknown Source)
at com.airtel.billreminder.config.BatchConfig.notificationJobLauncher(BatchConfig.java:343)
at sun.reflect.GeneratedMethodAccessor75.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.scheduling.support.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

我已经在这个问题上停留了2天了，我搜索了所有可能的文章或以前的答案，但是没有一个提示给我提示来处理单个处理器下指向不同数据库连接的2个实体管理器。

任何帮助，将不胜感激...

最佳答案

您正在处理器上使用@Transactional，它将在Spring Batch驱动的事务范围内执行。 Spring Batch为您处理事务管理，因此您无需使处理器或编写器具有事务性。我建议从处理器中删除此注释，并使其参与框架驱动的事务。

错误javax.persistence.TransactionRequiredException: no transaction is in progress表示未使用您的JpaTransactionManager。您需要：


通过返回JPA事务管理器，使批处理配置类之一扩展DefaultBatchConfigurer并覆盖getTransactionManager
或在您的上下文中添加类型为BatchConfigurer的bean，并使其返回JpaTransactionManager


这是一个例子：

@Bean
public BatchConfigurer batchConfigurer(EntityManagerFactory entityManagerFactory) {
   return new DefaultBatchConfigurer() {
       @Override
       public PlatformTransactionManager getTransactionManager() {
          return new JpaTransactionManager(entityManagerFactory);
       }
   };
}

请注意，这要求Spring Batch v4.1 +能够正常工作。