首页 文章

Spring Batch异步处理器配置可获得最佳性能

提问于
浏览
14

我在Spring Batch中创建异步处理器时遇到问题 . 我的处理器从 reader 获取 ID 并根据 SOAP 调用的响应创建对象 . 有时1输入( ID )必须有例如60-100 SOAP 调用,有时只是1.我试图制作多线程步骤,它正在处理例如50个输入,但它没用,因为49个线程在1秒内完成了他们的工作并被阻止,等待这个正在做60- 100 SOAP 来电 . 现在我使用 AsyncItemProcessor AsyncItemWriter 但这个解决方案对我来说很慢 . 由于我的输入( IDs )很大,从DB读取的大约25k项目我希望在时间开始~50-100输入 .

这是我的配置:

@Configuration
public class BatchConfig {

    @Autowired
    public JobBuilderFactory jobBuilderFactory;
    @Autowired
    public StepBuilderFactory stepBuilderFactory;
    @Autowired
    private DatabaseConfig databaseConfig;
    @Value(value = "classpath:Categories.txt")
    private Resource categories;

    @Bean
    public Job processJob() throws Exception {
        return jobBuilderFactory.get("processJob").incrementer(new RunIdIncrementer()).listener(listener()).flow(orderStep1()).end().build();
    }

    @Bean
    public Step orderStep1() throws Exception {
        return stepBuilderFactory.get("orderStep1").<Category, CategoryDailyResult>chunk(1).reader(reader()).processor(asyncItemProcessor()).writer(asyncItemWriter()).taskExecutor(taskExecutor()).build();
    }

    @Bean
    public JobExecutionListener listener() {
        return new JobCompletionListener();
    }


    @Bean
    public ItemWriter asyncItemWriter() {
        AsyncItemWriter<CategoryDailyResult> asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(itemWriter());
        return asyncItemWriter;
    }

    @Bean
    public ItemWriter<CategoryDailyResult> itemWriter(){
        return new Writer();
    }

    @Bean
    public ItemProcessor asyncItemProcessor() {
        AsyncItemProcessor<Category, CategoryDailyResult> asyncItemProcessor = new AsyncItemProcessor<>();
        asyncItemProcessor.setDelegate(itemProcessor());
        asyncItemProcessor.setTaskExecutor(taskExecutor());
        return asyncItemProcessor;
    }

    @Bean
    public ItemProcessor<Category, CategoryDailyResult> itemProcessor(){
        return new Processor();
    }

    @Bean
    public TaskExecutor taskExecutor(){
        SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
        taskExecutor.setConcurrencyLimit(50);
        return taskExecutor;
    }

    @Bean(destroyMethod = "")
    public ItemReader<Category> reader() throws Exception {
        String query = "select c from Category c where not exists elements(c.children)";

        JpaPagingItemReader<Category> reader = new JpaPagingItemReader<>();
        reader.setSaveState(false);
        reader.setQueryString(query);
        reader.setEntityManagerFactory(databaseConfig.entityManagerFactory().getObject());
        reader.setPageSize(1);

        return reader;
    }
}

如何提升我的申请?也许我做错了什么?任何反馈欢迎;)

@Edit:对于ID的输入:1到100我想要50个正在执行处理器的线程 . 我希望它们不会相互阻塞:Thread1进程输入“1”2分钟,此时我希望Thread2处理输入“2”,“8”,“64”这些很小并在几秒钟内执行 .

@ Edit2: My goal: 我在数据库中有25k ID,我用 JpaPagingItemReader 读取它们,每个ID都由处理器处理 . 每个项目彼此独立 . 对于每个ID,我在循环中调用 SOAP 调用0-100次然后创建对象,我将其传递给 Writer 并保存在数据库中 . 如何才能获得此类任务的最佳性能?

2 回答

  • 0

    你应该划分你的工作 . 像这样添加一个分区步骤:

    @Bean
    public Step partitionedOrderStep1(Step orderStep1) {
        return stepBuilder.get("partitionedOrderStep1")
                .partitioner(orderStep1)
                .partitioner("orderStep1", new SimplePartitioner())
                .taskExecutor(taskExecutor())
                .gridSize(10)  //Number of concurrent partitions
                .build();
    }
    

    然后在作业定义中使用该步骤 . .gridSize()调用配置要同时执行的分区数 . 如果您的任何Reader,Processor或Writer对象都是有状态的,则需要使用@StepScope对它们进行注释 .

  • 1

    @KCrookedHand:我已经处理了similar kind of scenario,我必须阅读几千个并且需要调用SOAP服务(我已将其注入到itemReader中)以获得匹配条件 .

    我的配置如下所示,基本上你有几个选项来实现并行处理,其中两个是'分区'和'客户端服务器'方法 . 我选择了分区,因为我可以根据数据更多地控制我需要多少个分区 .

    请提及@MichaelMinella提到的ThreadPoolTaskExecutor,以及适用于tasklet的下面的Step-Execution .

    <batch:step id="notificationMapper">
                <batch:partition partitioner="partitioner"
                    step="readXXXStep" />
            </batch:step>
        </batch:job>
    
    
        <batch:step id="readXXXStep">
            <batch:job ref="jobRef" job-launcher="jobLauncher"
                job-parameters-extractor="jobParameterExtractor" />
        </batch:step>
    
        <batch:job id="jobRef">
    
            <batch:step id="dummyStep" next="skippedItemsDecision">
                <batch:tasklet ref="dummyTasklet"/>
                <batch:listeners>
                    <batch:listener ref="stepListener" />
                </batch:listeners>
            </batch:step>
    
            <batch:step id="xxx.readItems" next="xxx.then.finish">
                <batch:tasklet>
                    <batch:chunk reader="xxxChunkReader" processor="chunkProcessor"
                        writer="itemWriter" commit-interval="100">
                    </batch:chunk>
                </batch:tasklet>
                <batch:listeners>
                    <batch:listener ref="taskletListener" />
                </batch:listeners>
            </batch:step>
    
            ...
    

相关问题