Можем ли мы обрабатывать несколько файлов последовательно, используя Spring Batch, в то время как несколько потоков используются для обработки данных отдельных файлов? - PullRequest
Новая
       57

Можем ли мы обрабатывать несколько файлов последовательно, используя Spring Batch, в то время как несколько потоков используются для обработки данных отдельных файлов?

1 голос
/ 27 марта 2020

Я хочу обрабатывать несколько файлов последовательно, и каждый файл должен обрабатываться с помощью нескольких потоков, поэтому я использовал подпружиненный пакет FlatFileItemReader и TaskExecutor, и, похоже, он работает нормально для меня. Как упоминалось в требовании, мы должны обрабатывать несколько файлов, поэтому наряду с FlatFileItemReader я использую MultiResourceItemReader, который будет принимать несколько файлов и обрабатывать один за другим, где у меня возникают проблемы. Может кто-нибудь помочь мне, что является причиной исключения? Какой подход это исправить ..? 

org.springframework.batch.item.ReaderNotOpenException: Reader must be open before it can be read.
   at org.springframework.batch.item.file.FlatFileItemReader.readLine(FlatFileItemReader.java:195) ~[spring-batch-infrastructure-3.0.5.RELEASE.jar:3.0.5.RELEASE]
   at org.springframework.batch.item.file.FlatFileItemReader.doRead(FlatFileItemReader.java:173) ~[spring-batch-infrastructure-3.0.5.RELEASE.jar:3.0.5.RELEASE]
   at org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader.read(AbstractItemCountingItemStreamItemReader.java:88) ~[spring-batch-infrastructure-3.0.5.RELEASE.jar:3.0.5.RELEASE]
   at org.springframework.batch.item.file.MultiResourceItemReader.readFromDelegate(MultiResourceItemReader.java:140) ~[spring-batch-infrastructure-3.0.5.RELEASE.jar:3.0.5.RELEASE]
   at org.springframework.batch.item.file.MultiResourceItemReader.readNextItem(MultiResourceItemReader.java:119)

customer2.csv 

200,Zoe,Nelson,1973-01-12 17:19:30
201,Vivian,Love,1951-10-31 08:57:08
202,Charde,Lang,1967-02-23 12:24:26

customer3.csv 

400,Amelia,Osborn,1972-05-09 09:21:22
401,Gemma,Finch,1989-09-25 23:00:59
402,Orli,Slater,1959-03-30 15:54:32
403,Donovan,Beasley,1986-06-18 14:50:30

customer4.csv 

600,Zelenia,Henson,1982-07-03 03:28:39
601,Thomas,Mathews,1954-11-21 20:34:03
602,Kevyn,Whitney,1984-09-21 06:24:25
603,Marny,Leon,1984-06-10 21:32:09
604,Jarrod,Gay,1960-06-22 19:11:04

customer5.csv 

800,Imogene,Lee,1966-10-19 17:53:44
801,Mira,Franks,1964-03-08 09:47:43
802,Silas,Dixon,1953-04-11 01:37:51
803,Paloma,Daniels,1962-06-14 17:01:02

Мой код: 

@Bean
public MultiResourceItemReader<Customer> multiResourceItemReader() {

    System.out.println("In multiResourceItemReader");
    MultiResourceItemReader<Customer> reader = new MultiResourceItemReader<>();
    reader.setDelegate(customerItemReader());
    reader.setResources(inputFiles);
    return reader;
}

@Bean
public FlatFileItemReader<Customer> customerItemReader() {
    FlatFileItemReader<Customer> reader = new FlatFileItemReader<>();
    DefaultLineMapper<Customer> customerLineMapper = new DefaultLineMapper<>();
    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
    tokenizer.setNames(new String[] {"id", "firstName", "lastName", "birthdate"});
    customerLineMapper.setLineTokenizer(tokenizer);
    customerLineMapper.setFieldSetMapper(new CustomerFieldSetMapper());
    customerLineMapper.afterPropertiesSet();
    reader.setLineMapper(customerLineMapper);
    return reader;
}

сильфон работает нормально при использовании ниже: 

@Bean
public Step step1() {
    return stepBuilderFactory.get("step1")
            .<Customer, Customer>chunk(100).
            reader(customerItemReader())
            .writer(customerItemWriter()).taskExecutor(taskExecutor()).throttleLimit(10)
            .build();
}
}

сильфон не работает работая над вышеупомянутым исключением 

@Bean
public Step step1() {
    return stepBuilderFactory.get("step1")
            .<Customer, Customer>chunk(100).
            reader(multiResourceItemReader())
            .writer(customerItemWriter()).taskExecutor(taskExecutor()).throttleLimit(10)
            .build();
}

1 Ответ

0 голосов
/ 30 марта 2020

Поскольку вы используете средство чтения в многопоточном шаге, поток мог закрыть текущий файл, в то время как другой поток пытается прочитать из этого файла одновременно. Вам необходимо синхронизировать доступ к вашему считывателю с SynchronizedItemStreamReader: 

@Bean
public SynchronizedItemStreamReader<Customer> multiResourceItemReader() {
    System.out.println("In multiResourceItemReader");
    MultiResourceItemReader<Customer> reader = new MultiResourceItemReader<>();
    reader.setDelegate(customerItemReader());
    reader.setResources(inputFiles);

    SynchronizedItemStreamReader<Customer> synchronizedItemStreamReader = new SynchronizedItemStreamReader<>();
    synchronizedItemStreamReader.setDelegate(reader);
    return synchronizedItemStreamReader;
}
...