import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
ItemReader<Map<String, Object>> hdfsFileReader(HdfsResourceLoader resourceLoader, LineMapper<Map<String, Object>> lineMapper) throws IOException {
	MultiResourceItemReader<Map<String, Object>> multiReader = new MultiResourceItemReader<>();
	Resource[] resources = resourceLoader.getResources("/demo/influencers/*");
	multiReader.setResources(resources);	
	FlatFileItemReader<Map<String, Object>> itemReader = new FlatFileItemReader<>();
	itemReader.setLineMapper(lineMapper);
	multiReader.setDelegate(itemReader);
	return multiReader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
@Qualifier(STEP_NAME)
public FlatFileItemReader<InnofactorImportFileLine> innofactorReader(
        @Value("#{jobParameters['inputFile']}") String inputFile) {

    FlatFileItemReader<InnofactorImportFileLine> reader = new FlatFileItemReader<>();

    reader.setEncoding(StandardCharsets.UTF_8.name());
    reader.setLineMapper(innofactorImportLineMapper());
    reader.setStrict(true);
    reader.setResource(new FileSystemResource(inputFile));
    reader.setLinesToSkip(0);
    reader.setBufferedReaderFactory(bufferedReaderFactory);

    return reader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public JdbcBatchItemWriter<Customer> customerItemWriter() {
	JdbcBatchItemWriter<Customer> itemWriter = new JdbcBatchItemWriter<>();

	itemWriter.setDataSource(this.dataSource);
	itemWriter.setSql("INSERT INTO NEW_CUSTOMER VALUES (:id, :firstName, :lastName, :birthdate)");
	itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider());
	itemWriter.afterPropertiesSet();

	return itemWriter;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
protected OrderImportItemReader orderReader(@Value("#{jobParameters['resource']}") final Resource orderCsvResource) {

    FlatFileItemReader flatFileItemReader = new FlatFileItemReader<>();
    flatFileItemReader.setLineMapper(new DefaultLineMapper<OrderCsvLineValue>() {{
        setLineTokenizer(new DelimitedLineTokenizer() {{
            setNames(ORDER_CSV_HEADER_NAMES);

        }});
        setFieldSetMapper(new BeanWrapperFieldSetMapper<OrderCsvLineValue>() {{
            setTargetType(OrderCsvLineValue.class);
        }});
    }});
    flatFileItemReader.setLinesToSkip(1);
    flatFileItemReader.setResource(orderCsvResource);

    SingleItemPeekableItemReader singleItemPeekableItemReader = new SingleItemPeekableItemReader();
    singleItemPeekableItemReader.setDelegate(flatFileItemReader);

    OrderImportItemReader reader = new OrderImportItemReader();
    reader.setDelegate(singleItemPeekableItemReader);

    return reader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
/**
 * @return reader of signatures from transcript files in inbox directory
 */
@Bean
@StepScope
public MultiResourceItemReader<Signature> transcriptSignatureItemReader() {

    /* Get transcript resources */
    Resource[] resources = listIdsStore()
        .stream()
        . <Resource> map(listId ->
                new FileSystemResource(INBOX_DIR + listId + ".txt"))
        .toArray(size -> new Resource[size]);

    /* Create MultiResourceItemReader */
    MultiResourceItemReader<Signature> multiResourceReader =
            new MultiResourceItemReader<>();
    multiResourceReader.setResources(resources);
    multiResourceReader.setDelegate(trascriptSignatureReader());

    return multiResourceReader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
FlatFileItemReader<Person> flatFileItemReader(@Value("#{jobParameters[file]}") File file) {
    FlatFileItemReader<Person> r = new FlatFileItemReader<>();
    r.setResource(new FileSystemResource(file));
    r.setLineMapper(new DefaultLineMapper<Person>() {
        {
            this.setLineTokenizer(new DelimitedLineTokenizer(",") {
                {
                    this.setNames(new String[]{"first", "last", "email"});
                }
            });
            this.setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {
                {
                    this.setTargetType(Person.class);
                }
            });
        }
    });
    return r;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean(name = "sendSystemMessageJobUserAnonymizedProcessor", destroyMethod = "")
@StepScope
private ItemProcessor<UUID, MessageDestination> userAnonymizedProcessor()
{
	return new ItemProcessor<UUID, MessageDestination>() {
		@Value("#{jobParameters['messageText']}")
		private String messageText;

		@Override
		public MessageDestination process(UUID userAnonymizedId) throws Exception
		{
			logger.debug("Processing user anonymized with id {}", userAnonymizedId);
			UserAnonymizedDto userAnonymized = userAnonymizedService.getUserAnonymized(userAnonymizedId);

			MessageDestination messageDestination = messageDestinationRepository
					.findOne(userAnonymized.getAnonymousDestination().getId());
			messageDestination.send(SystemMessage.createInstance(messageText));

			return messageDestination;
		}
	};
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public ItemReader<String> itemReader() {
	final List<String> items = IntStream.range(0, 1000).mapToObj(String::valueOf).collect(toList());

	return new ListItemReader<String>(items);
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
/**
 * maps CSV data into rows of fields, which are then
 * mapped to Customer.class instances based on conventions:
 * <p/>
 * <CODE>col 1 => firstName => customer.setFirstName(String)</CODE>
 */
@Bean(name = readCsvFileIntoTableStepReader)
@StepScope
public FlatFileItemReader<Customer> reader(@Value("#{jobParameters['input.file']}") Resource resource) throws Exception {

    log.debug(String.format("building FlatFileItemReader to read in the file %s", resource.getFile().getAbsolutePath()));

    FlatFileItemReader<Customer> csvFileReader = new FlatFileItemReader<Customer>();
    csvFileReader.setResource(resource);

    DelimitedLineTokenizer delimitedLineTokenizer = new DelimitedLineTokenizer(DelimitedLineTokenizer.DELIMITER_COMMA);
    delimitedLineTokenizer.setNames(new String[]{"lastName", "firstName"});

    BeanWrapperFieldSetMapper<Customer> beanWrapperFieldSetMapper = new BeanWrapperFieldSetMapper<Customer>();
    beanWrapperFieldSetMapper.setTargetType(Customer.class);

    DefaultLineMapper<Customer> defaultLineMapper = new DefaultLineMapper<Customer>();
    defaultLineMapper.setLineTokenizer(delimitedLineTokenizer);
    defaultLineMapper.setFieldSetMapper(beanWrapperFieldSetMapper);

    csvFileReader.setLineMapper(defaultLineMapper);
    return csvFileReader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public FlatFileItemReader<Product> productReader(@Value("#{jobParameters['inputFile']}") String inputFile) {

	DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer(",");
	tokenizer.setNames(new String[] { "PRODUCT_ID", "NAME", "DESCRIPTION", "PRICE" });

	DefaultLineMapper<Product> lineMapper = new DefaultLineMapper<Product>();
	lineMapper.setLineTokenizer(tokenizer);
	lineMapper.setFieldSetMapper(new ProductFieldSetMapper());

	FlatFileItemReader<Product> reader = new FlatFileItemReader<Product>();

	try {
		reader.setResource(new FileSystemResource(ResourceUtils.getFile(inputFile)));
	} catch (FileNotFoundException e) {
		log.error("파일을 찾을 수 없습니다. inputFile=" + inputFile, e);
		throw new RuntimeException(e);
	}
	reader.setLinesToSkip(1);
	reader.setLineMapper(lineMapper);

	return reader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
@Qualifier(IMPORT_STEP)
public FlatFileItemReader<LHHuntingClubCSVRow> lhClubImportReader(
        @Value("#{jobExecutionContext['inputFile']}") String inputFile) {
    final FlatFileItemReader<LHHuntingClubCSVRow> reader = new FlatFileItemReader<>();

    reader.setEncoding(StandardCharsets.ISO_8859_1.name());
    reader.setLineMapper(inputLineMapper());
    reader.setStrict(true);
    reader.setResource(new FileSystemResource(inputFile));
    reader.setLinesToSkip(1);

    return reader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
@Qualifier(STEP_NAME)
public MetsastajaRekisteriPersonItemWriter innofactorWriter(
        @Value("#{jobParameters['importTimestamp']}") long importTimestamp) {
    return new MetsastajaRekisteriPersonItemWriter(importService, importTimestamp);
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public JdbcPagingItemReader<Customer> pagingItemReader(
		@Value("#{stepExecutionContext['minValue']}")Long minValue,
		@Value("#{stepExecutionContext['maxValue']}")Long maxValue) {
	System.out.println("reading " + minValue + " to " + maxValue);
	JdbcPagingItemReader<Customer> reader = new JdbcPagingItemReader<>();

	reader.setDataSource(this.dataSource);
	reader.setFetchSize(1000);
	reader.setRowMapper(new CustomerRowMapper());

	MySqlPagingQueryProvider queryProvider = new MySqlPagingQueryProvider();
	queryProvider.setSelectClause("id, firstName, lastName, birthdate");
	queryProvider.setFromClause("from customer");
	queryProvider.setWhereClause("where id >= " + minValue + " and id <= " + maxValue);

	Map<String, Order> sortKeys = new HashMap<>(1);

	sortKeys.put("id", Order.ASCENDING);

	queryProvider.setSortKeys(sortKeys);

	reader.setQueryProvider(queryProvider);

	return reader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public ItemReader<String> firstReader(@Value("#{jobParameters[message]}") String text) {
    log.info("+++ r-r-r  {} ", text);
  return new ListItemReader<>(Lists.newArrayList(text));
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public ItemProcessor<Path, NamedBytes> processor(@Value("#{jobParameters[globalParams]}") String globalParams) {
    CompositeItemProcessor<Path, NamedBytes> processors = new CompositeItemProcessor<>();
    FileContentItemProcessor fileContentProcessor = new FileContentItemProcessor();
    GlobalParametersEnricherProcessor paramsEnricherProcessor = new GlobalParametersEnricherProcessor(globalParams);
    FileExtensionStripperProcessor fileExtensionStripperProcessor = new FileExtensionStripperProcessor();
    PlantUmlItemProcessor plantUmlProcessor = new PlantUmlItemProcessor(settings);
    processors.setDelegates(Arrays.asList(fileContentProcessor, paramsEnricherProcessor, fileExtensionStripperProcessor,
        plantUmlProcessor));
    return processors;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean(name = "activityAggregationJobDayActivityReader", destroyMethod = "")
@StepScope
public ItemReader<Long> dayActivityReader()
{
	return intervalActivityIdReader(Date.valueOf(
			TimeUtil.getStartOfDay(DEFAULT_TIME_ZONE, ZonedDateTime.now(DEFAULT_TIME_ZONE)).minusDays(1).toLocalDate()),
			DayActivity.class, DAY_ACTIVITY_CHUNK_SIZE);
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean(name = "activityAggregationJobWeekActivityReader", destroyMethod = "")
@StepScope
public ItemReader<Long> weekActivityReader()
{
	return intervalActivityIdReader(Date.valueOf(
			TimeUtil.getStartOfWeek(DEFAULT_TIME_ZONE, ZonedDateTime.now(DEFAULT_TIME_ZONE)).minusWeeks(1).toLocalDate()),
			WeekActivity.class, WEEK_ACTIVITY_CHUNK_SIZE);
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean
@StepScope
public Tasklet workerTasklet(
		final @Value("#{stepExecutionContext['partitionNumber']}")Integer partitionNumber) {

	return new Tasklet() {
		@Override
		public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
			System.out.println("This tasklet ran partition: " + partitionNumber);

			return RepeatStatus.FINISHED;
		}
	};
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean(destroyMethod = "")
@StepScope
public JpaPagingItemReader<Employee> taxCalculatorItemReader(@Value("#{stepExecution}") StepExecution stepExecution) {
    JpaPagingItemReader<Employee> employeeItemReader = new JpaPagingItemReader<>();

    employeeItemReader.setEntityManagerFactory(persistenceConfig.entityManagerFactory());
    employeeItemReader.setQueryString(TaxCalculation.GET_UNPROCESSED_EMPLOYEES_BY_YEAR_AND_MONTH_QUERY);
    Map<String, Object> parameters = new HashMap<>();
    parameters.put("year", stepExecution.getJobParameters().getLong("year").intValue());
    parameters.put("month", stepExecution.getJobParameters().getLong("month").intValue());
    parameters.put("jobExecutionId", stepExecution.getJobExecutionId());
    employeeItemReader.setParameterValues(parameters);
    employeeItemReader.setSaveState(false);
    return employeeItemReader;
}
 

import org.springframework.batch.core.configuration.annotation.StepScope; //导入依赖的package包/类
@Bean(destroyMethod = "")
@StepScope
public JpaPagingItemReader<Employee> taxCalculatorItemReaderSlave(@Value("#{stepExecution}") StepExecution stepExecution) {
    JpaPagingItemReader<Employee> employeeItemReader = new JpaPagingItemReader<>();
    employeeItemReader.setEntityManagerFactory(persistenceConfig.entityManagerFactory());
    employeeItemReader.setQueryString(TaxCalculation.GET_UNPROCESSED_EMPLOYEES_BY_YEAR_AND_MONTH_QUERY_SLAVE);
    Map<String, Object> parameters = new HashMap<>();
    parameters.put("year", stepExecution.getJobParameters().getLong("year").intValue());
    parameters.put("month", stepExecution.getJobParameters().getLong("month").intValue());
    parameters.put("jobExecutionId", stepExecution.getJobExecutionId());
    parameters.put("minId", stepExecution.getExecutionContext().getLong("minValue"));
    parameters.put("maxId", stepExecution.getExecutionContext().getLong("maxValue"));
    employeeItemReader.setParameterValues(parameters);
    return employeeItemReader;
}