问春季批量向Kafka写入数据

EN

@Autowired
private JobBuilderFactory jobBuilderFactory;

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Autowired
@Qualifier("flatFileWriter")
private ItemWriter<? super Demo1> flatFileWriter;
@Bean
public Job FileReaderJob() {
    return jobBuilderFactory.get("FileReaderJob").start(FileReaderStep()).build();
}
private Step FileReaderStep() {
    // TODO Auto-generated method stub
    return stepBuilderFactory.get("FileReaderStep").<Demo1,Demo1>chunk(100).reader(flatFileReader())
            .writer(flatFileWriter).build();
}

@Bean
@StepScope
public FlatFileItemReader<Demo1> flatFileReader() {
    // TODO Auto-generated method stub

    FlatFileItemReader<Demo1> reader = new FlatFileItemReader<Demo1>();
    reader.setResource(new  ClassPathResource("Demo1.csv"));
    reader.setLinesToSkip(1);
    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
    tokenizer.setNames(new String [] {"id","first","last"});
    DefaultLineMapper<Demo1> mapper = new DefaultLineMapper<>();
    mapper.setLineTokenizer(tokenizer);
    mapper.setFieldSetMapper(new FieldSetMapper<Demo1>() {

        @Override
        public Demo1 mapFieldSet(FieldSet fieldSet) throws BindException {
            Demo1 demo1 = new Demo1();
            demo1.setId(fieldSet.readLong("id"));
            demo1.setFirst(fieldSet.readString("first"));
            demo1.setLast(fieldSet.readString("last"));

            // TODO Auto-generated method stub
            return demo1;
        }
    });

    mapper.afterPropertiesSet();
    reader.setLineMapper(mapper);

    return reader;
}