Пакетная загрузка Spring @BeforeStep не работает в CompositeItemWriter - PullRequest
Новая
       27

Пакетная загрузка Spring @BeforeStep не работает в CompositeItemWriter

1 голос
/ 02 августа 2020

Перед написанием этого запроса я просмотрел много ссылок, например: Spring-batch @BeforeStep не работает с @ StepScope и Проблема с объемом пакета Spring при использовании Spring boot , но все же невозможно заставить это работать.

Похоже, что с приведенными ниже конфигурациями @BeforeStep не выполняется, и я не понимаю root причину проблемы. Какие-нибудь быстрые указатели?

Похоже, Spring Boot 2.2.7.RELEASE (или Spring Boot 2 ) имеет проблему с @StepScope, и даже это не работает, когда у вас CompositeItemWriter . Эта проблема уже решена? Даже добавление @Bean ниже не помогает. 

@Bean
public StepScope stepScope() {
    final StepScope stepScope = new StepScope();
    stepScope.setAutoProxy(true);
    return stepScope;
}

Примечание. Здесь я использую CompositeItemWriter

@Configuration
public class MyJob {

    @Value("${skip.limit}")
    private Integer skipLimit;

    @Value("${chunk.size}")
    private Integer chunkSize;

    @Bean("employeeJob")
    public Job readEmployeejob(JobBuilderFactory jobBuilderFactory, 
            StepBuilderFactory stepBuilderFactory,
            JdbcCursorItemReader<Employee> employeeItemReader) throws Exception {

        return jobBuilderFactory.get("employeeJob")
                .incrementer(new RunIdIncrementer())
                .start(abcStep(stepBuilderFactory, employeeItemReader))
                .next(.....)
                .next(.....)
                .next(.....)
                .next(.....)
                .start(xyzStep(stepBuilderFactory, abcReder))
                .listener(abcListener())
                .build();
    }

    @Bean
    public Step abcStep(StepBuilderFactory stepBuilderFactory,
            JdbcCursorItemReader<Employee> employeeItemReader) {
        return stepBuilderFactory.get("abcStep").
                <Employee, Employee>chunk(chunkSize)
                .reader(employeeItemReader)
                .processor(employeeProcessor())
                .writer(employeeCompositeWriter())
                .faultTolerant()
                .skip(Exception.class)
                .skipLimit(skipLimit)
                .listener(abcSkipListener())
                .listener(abcStepListener())
                .build();
    }

    @Bean
    public EmployeeProcessor employeeProcessor() {
        return new EmployeeProcessor();
    }

    @Bean
    public CompositeItemWriter<Employee> employeeCompositeWriter() throws Exception {
        List<ItemWriter<? super Employee>> e = new ArrayList<>();
        e.add(EmployeeWriter());
        e.add(..........);

        CompositeItemWriter<Employee> compositeItemWriter = new CompositeItemWriter<>();
        compositeItemWriter.setDelegates(e);
        compositeItemWriter.afterPropertiesSet();
        return compositeItemWriter;
    }
    ...
    ...
    ...
    // All other @Beans 
}

Writer 

public class EmployeeWriter implements ItemWriter<Employee> {
    @Autowired
    private NamedParameterJdbcTemplate namedJdbcTemplate;
    
    private Long jobId;
    private StepExecution stepExecution; 

    private static final String SQL = "INSERT QUERY"

    @Override
    public void write(List<? extends Employee> items) throws Exception {
        if(stepExecution.getJobExecution().getExecutionContext() != null) {
            long count=(long) stepExecution.getJobExecution().getExecutionContext().get(MDSConst.VALID_PROCS_PDF_COUNT);
            count=count+items.size();
            stepExecution.getJobExecution().getExecutionContext().put(MDSConst.VALID_PROCS_PDF_COUNT, count);               
        }
    
    }

    @SuppressWarnings("unchecked")
    private void saveEmployee(List<? extends Employee> items) {
        List<Map<String, Object>> batchValues = new ArrayList<>(items.size());

        for (Employee acct : items) {
            batchValues.add(new MapSqlParameterSource()
                    .addValue("empId", acct.getRecType())
                    ......
                    ......
                    ......
                    ......
                    ......
                    ......
                    ......
                    ......
                    .getValues());
        }

        try {
            int[] updateCounts = namedJdbcTemplate.batchUpdate(SQL, batchValues.toArray(new Map[items.size()]));
        } catch (Exception e) {
            log.error("Batch Insert Error for ", e.getMessage());
        }
    }
    
    @BeforeStep
    public void getInterstepData(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
        this.jobId = stepExecution.getJobExecution().getJobId();
    }
}
...