如何正确地将syncitemwriter添加到一个步骤中？

Question

在春季批处理作业中，如下所示，我尝试使用asyncwriter，

@Bean
public Step readWriteStep() throws Exception {
    return stepBuilderFactory.get("readWriteStep")
        .listener(listener)
        .<Data, Data>chunk(10)
        .reader(dataItemReader())
        .writer(dataAsyncWriter())
        .build();
}

@Bean
public AsyncItemWriter<Data> dataAsyncWriter() throws Exception {
    AsyncItemWriter<Data> asyncItemWriter = new AsyncItemWriter<>();
    asyncItemWriter.setDelegate(dataItemWriter);
    asyncItemWriter.afterPropertiesSet();
    return asyncItemWriter;
}

如果我尝试这样的Intellij抱怨：

Required type: ItemWriter <? super Data>
Provided: AsyncItemWriter <Data>

当我更改。data，data，data＆gt; chunk（10） to 时。 ＆lt; data，future＆lt; data＆gt;＆gt;块（10） Intellij不会发出任何警告，但是当我运行作业时，我会得到以下例外：

java.lang.ClassCastException: Data cannot be cast to class java.util.concurrent.Future Data is in unnamed module of loader 'app'; 
java.util.concurrent.Future is in module java.base of loader 'bootstrap'

对于第一个和第二个参数是什么？ 。＆lt; data，data＆gt;块（10）？

这两个参数是处理器所需的两个参数，第二个参数是处理器回馈什么？

如何解决这个问题？

Answer 1

如果您更改步骤定义以使用以下内容，则应编译您的示例：

.<Data, Future<Data>>chunk(10)

也就是说，我不确定在运行时可以正常工作，因为asyncitemWriter有望从其封闭式中解开构造项目未来，其中这些未来 s由 asyncitempropersorsor 。

换句话说，asyncitemWriter和asyncitemProcessor应结合使用此模式来使用。这是两者的快速示例：

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Future;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.integration.async.AsyncItemProcessor;
import org.springframework.batch.integration.async.AsyncItemWriter;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

@Configuration
@EnableBatchProcessing
public class SO72477556 {

    @Bean
    public ItemReader<Data> dataItemReader() {
        return new ListItemReader<Data>(Arrays.asList());
    }


    @Bean
    public ItemProcessor<Data, Data> dataItemProcessor() {
        return new ItemProcessor<Data, Data>() {
            @Override
            public Data process(Data item) throws Exception {
                return item;
            }
        };
    }

    @Bean
    public AsyncItemProcessor<Data, Data> asyncDataItemProcessor() {
        AsyncItemProcessor<Data, Data> asyncItemProcessor = new AsyncItemProcessor<>();
        asyncItemProcessor.setDelegate(dataItemProcessor());
        asyncItemProcessor.setTaskExecutor(new SimpleAsyncTaskExecutor());
        return asyncItemProcessor;
    }

    @Bean
    public ItemWriter<Data> dataItemWriter() {
        return new ItemWriter<Data>() {
            @Override
            public void write(List<? extends Data> items) throws Exception {

            }
        };
    }

    @Bean
    public AsyncItemWriter<Data> dataAsyncWriter() throws Exception {
        AsyncItemWriter<Data> asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(dataItemWriter());
        asyncItemWriter.afterPropertiesSet();
        return asyncItemWriter;
    }

    @Bean
    public Step readWriteStep(StepBuilderFactory stepBuilderFactory) throws Exception {
        return stepBuilderFactory.get("readWriteStep")
                .<Data, Future<Data>>chunk(10)
                .reader(dataItemReader())
                .processor(asyncDataItemProcessor())
                .writer(dataAsyncWriter())
                .build();
    }

    @Bean
    public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) throws Exception {
        return jobs.get("job")
                .start(readWriteStep(steps))
                .build();
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(SO72477556.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

    static class Data {}

}