标签: spring-batch

我特别对Spring和Spring-Batch很新.我还是设法安装了Spring Batch-Admin.我添加了自定义作业和Hibernate/JPA以实现持久性.

一切都按预期工作,直到第一个块应该持久化.然后我收到以下错误消息:

org.springframework.transaction.CannotCreateTransactionException: 
      Could not open JPA  EntityManager for transaction;

nested exception is java.lang.IllegalStateException: Already value
      [org.springframework.jdbc.datasource.ConnectionHolder@60d31437] 
      for key [org.springframework.jdbc.datasource.DriverManagerDataSource@12da4b19] 
      bound to thread [jobLauncherTaskExecutor-1]

这是完整的堆栈跟踪:

org.springframework.transaction.CannotCreateTransactionException: Could not open JPA  EntityManager for transaction; nested exception is java.lang.IllegalStateException: Already value [org.springframework.jdbc.datasource.ConnectionHolder@43f9e588] for key [org.springframework.jdbc.datasource.DriverManagerDataSource@84f171a] bound to thread [jobLauncherTaskExecutor-1]
     at org.springframework.orm.jpa.JpaTransactionManager.doBegin(JpaTransactionManager.java:427)
     at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:371)
     at org.springframework.transaction.interceptor.TransactionAspectSupport.createTransactionIfNecessary(TransactionAspectSupport.java:335)
     at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:105)
     at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
     at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
     at com.sun.proxy.$Proxy41.saveIfUnique(Unknown Source)
     at com.qompa.batch.ArticleItemWriter.write(ArticleItemWriter.java:28)
     at org.springframework.batch.core.step.item.SimpleChunkProcessor.writeItems(SimpleChunkProcessor.java:171)
     at org.springframework.batch.core.step.item.SimpleChunkProcessor.doWrite(SimpleChunkProcessor.java:150)
     at org.springframework.batch.core.step.item.FaultTolerantChunkProcessor$3.doWithRetry(FaultTolerantChunkProcessor.java:313)
     at org.springframework.batch.retry.support.RetryTemplate.doExecute(RetryTemplate.java:240)
     at …

我已经成功建立了一个Spring Batch项目教程.我真的很想知道是否可以在"Spring级别"使其成为多线程.

我想要的基本思路是制作一个任务列表或任务步骤,让它们被独立的线程拾取和处理,理想情况下是一个限于'n'个线程的池.

这可能吗？如果是这样,怎么样？有人可以从我目前所在的地方指导我到那一点吗？

简单的项目我是从本教程在这里.它基本上有不同的任务,可以将信息打印到屏幕上.

这是我当前的simpleJob.xml文件,其中包含作业详细信息:

<import resource="applicationContext.xml"/>

    <bean id="hello" class="helloworld.PrintTasklet">
        <property name="message" value="Hello"/>
    </bean>

    <bean id="space" class="helloworld.PrintTasklet">
        <property name="message" value=" "/>
    </bean>

    <bean id="world" class="helloworld.PrintTasklet">
        <property name="message" value="World!\n"/>
    </bean>

    <bean id="taskletStep" class="org.springframework.batch.core.step.tasklet.TaskletStep" >
        <property name="jobRepository" ref="jobRepository"/>
        <property name="transactionManager" ref="transactionManager"/>
    </bean>

    <bean id="simpleJob" class="org.springframework.batch.core.job.SimpleJob">
        <property name="name" value="simpleJob" />
        <property name="steps">
            <list>
                <bean parent="taskletStep">
                    <property name="tasklet" ref="hello"/>
                </bean>
                <bean parent="taskletStep">
                    <property name="tasklet" ref="space"/>
                </bean>
                <bean parent="taskletStep">
                    <property name="tasklet" ref="world"/>
                </bean>
            </list>
        </property>
        <property name="jobRepository" ref="jobRepository"/>
    </bean>

我的appContext包含作业存储库bean(SimpleJobRepository),事务管理器( …

Spring Batch的ItemWriter接口是这样的:

write(List<? extends T> items); 

我想让ItemWriter调用一个服务但我的服务有这个:

process(List<T> items); 

AFAIK,Java Generics对集合中的类型转换非常严格.

我希望我的春季批处理工作每3小时运行一次

我使用了表达式,* * */3 * * ?这可以在可被3整除的小时开始工作,例如说服务器在下午2点开始工作,工作仅在下午3点开始执行 - 到目前为止这么好,但工作每秒都开始工作!是因为我在第一个位置使用了*吗？

我试过0 0 */3 * * ?但它错了.实现这一目标的最佳方法是什么？

我有一个直接的要求,其中,我需要读取项目列表(从DB)并需要处理项目,一旦处理,它必须更新到DB.

我正在考虑将Spring批量块用于阅读器,处理器和写入器.我的读者将从列表中一次返回一个项目并将其发送到处理器,一旦处理结束,它将返回到Writer,它将更新数据库

我可能会稍后多线程处理这些方法中的一些同步成本.

在这里,我预见到一些问题.

1. 要处理的项目数可能更多.可能在10,000s甚至更多.
2. 处理器中需要一些逻辑计算.因此,一次处理1个项目.即使它是10个线程的多线程,也不确定性能.
3. Writer可以更新该处理项目的DB中的结果.不确定如何进行批量更新,因为它始终只处理了1个项目并准备就绪.

这种方法对于这种用例是否正确或者可以做得更好？还有一种方法可以在读者,处理器和写入器的一次调用中处理一堆物品吗？如果是这样,我是否需要创建一些机制,我从列表中提取10个项目并将其提供给处理器？似乎编写器会更新每个记录,只有当编写器收到一堆处理过的项目时,批量更新才有意义.有什么建议吗？

请为此设计注一些亮点以获得更好的性能.

谢谢,

我有如下配置:

batch:job id="reconciliationJob" job-repository="jobRepository" restartable="true"

在应用程序上下文启动期间,我在日志中收到类似的内容:

[INFO] [] []覆盖bean'reconciliationJob'的bean定义:替换[Generic bean:class [org.springframework.batch.core.configuration.xml.SimpleFlowFactoryBean]; 范围=; 抽象= FALSE; lazyInit = FALSE; autowireMode = 0; dependencyCheck = 0; autowireCandidate = TRUE; 初级= FALSE; factoryBeanName = NULL; factoryMethodName = NULL; initMethodName = NULL; 使用[Generic bean:class [org.springframework.batch.core.configuration.xml.JobParserJobFactoryBean]; destroyMethodName = null]; 范围=; 抽象= FALSE; lazyInit = FALSE; autowireMode = 0; dependencyCheck = 0; autowireCandidate = TRUE; 初级= FALSE; factoryBeanName = NULL; factoryMethodName = NULL; initMethodName = NULL; destroyMethodName =空]

我怎样才能解决这个最重要的问题？

我正在尝试从控制器运行我的批处理作业.它将由cron作业或访问特定链接启动.我正在使用Spring启动,没有XML只是注释.在我当前的设置中,我有一个包含以下bean的服务:

@EnableBatchProcessing
@PersistenceContext
public class batchService{

@Bean
public ItemReader<Somemodel> reader() {
...
}

@Bean
    public ItemProcessor<Somemodel, Somemodel> processor() {
        return new SomemodelProcessor();
    }
@Bean
    public ItemWriter writer() {
        CustomItemWriter writer = new CustomItemWriter();
            return writer; 
    }
@Bean
    public Job importUserJob(JobBuilderFactory jobs, Step s1) {
        return jobs.get("importUserJob")
                .incrementer(new RunIdIncrementer())
                .flow(s1)
                .end()
                .build();
    }

@Bean
public Step step1(StepBuilderFactory stepBuilderFactory,       
ItemReader<somemodel> reader,
            ItemWriter<somemodel> writer, ItemProcessor<somemodel,   
somemodel> processor) {
        return stepBuilderFactory.get("step1")
                .<somemodel, somemodel> chunk(100)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }   
}   

只要将@Configuration注释放在我的batchService类之上,作业就会在我运行应用程序后立即启动.它成功完成,一切都很好.现在我试图删除@Configuration注释并随时运行它.有没有办法从控制器发射它？

谢谢!

现在,我正在使用jobParameters获取我的FlatFileItemReader和FlatFileItemWriter的文件名.可以测试我的批处理,但我的目标是读取某个目录中的文件(此目录中只有此文件),文件名可能会更改.输出文件名应取决于输入文件名.

因此,我考虑在我的工作中添加一个新步骤,此步骤将通过搜索好目录并在其中查找文件来设置输出和输入文件名.我从Spring文档中读取了将数据传递给Future Steps,以及来自SO的这个线程,但是我无法使其工作,文件总是"空".

首先,我定义了以下Tasklet

public class SettingFilenamesTasklet implements Tasklet {

    private StepExecution stepExecution;

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        // TODO Search folder and set real filenames
        String inputFilename = "D:/TestInputFolder/dataFile.csv";
        String outputFilename = "D:/TestOutputFolder/dataFile-processed.csv";
        ExecutionContext stepContext = stepExecution.getExecutionContext();
        stepContext.put("inputFile", inputFilename);
        stepContext.put("outputFile", outputFilename);
        return RepeatStatus.FINISHED;
    }

    @BeforeStep
    public void saveStepExecution(StepExecution stepExec) {
        stepExecution = stepExec;
    }
}

然后,我添加了promotionListener bean

@Bean
public ExecutionContextPromotionListener promotionListener() {
    ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
    listener.setKeys(new String[]{
            "inputFile", …

我做了一些搜索,但找不到任何样本/示例.

我需要读取一个表(输入)的地理坐标,处理它以生成与坐标相关的POI.因此,一个地理坐标将导致需要将一个或多个POI插入另一个表(输出).

我目前正在使用JdbcCursorItemReader和JdbcBatchItemWriter来读取一个项目/记录并写入一个项目/记录.还有一个ItemProcessor,它为给定的地理坐标生成POI.

自定义JdbcBatchItemWriter是否帮助我实现了这一目标？

有任何想法吗？TIA.

我的问题涉及Spring批处理和事务.

假设我为我的一个步骤选择了50的提交间隔.

还假设我总共有1000条记录,其中一条记录会导致itemWriter失败,从而导致整个块的回滚(在我的例子中为50条记录).

什么是stategies以确保在作业完成后将49个有效记录写入数据库(并忽略有问题的块)？