java - 从 Spring Batch 自定义 ItemReader 返回对象列表

Question

我需要处理数千笔付款。所以我在 Spring Batch 中使用 VtdXml 而不是 StaxEventItemReader，我为此创建了自定义项目读取器。为了使用多线程读取巨大的 xml，我创建了 10 个线程的分区。我将巨大的 xml 文件分成 10 个文件并分配给分区中的每个线程。一旦我读取了 xml，我将转换为对象列表并发送给 Writer。在 Writer 中收到后，我将自定义对象列表并合并到最终列表中。每当我返回读取的对象列表时再次调用，它永远不会结束。如何将对象列表传递给编写器并合并到最终列表中？

public class VtdWholeItemReader<T> implements ResourceAwareItemReaderItemStream<T> {


private Resource resource;

private boolean noInput;

private boolean strict = true;

private InputStream inputStream;

private int index = 0;





@Override
public void open(ExecutionContext executionContext) {
    Assert.notNull(resource, "The Resource must not be null.");

    noInput = true;
    if (!resource.exists()) {
        if (strict) {
            throw new IllegalStateException("Input resource must exist (reader is in 'strict' mode)");
        }
        log.warn("Input resource does not exist " + resource.getDescription());
        return;
    }
    if (!resource.isReadable()) {
        if (strict) {
            throw new IllegalStateException("Input resource must be readable (reader is in 'strict' mode)");
        }
        log.warn("Input resource is not readable " + resource.getDescription());
        return;
    }
    noInput = false;
}

@Override
public void update(ExecutionContext executionContext) {

}

@Override
public void close() {
    try {
        if (inputStream != null) {
            inputStream.close();
        }
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } finally {
        inputStream = null;
    }

}

@Override
public void setResource(Resource resource) {
    this.resource = resource;
}

@Override
public T read()
        throws java.lang.Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
    if (noInput) {
        return null;
    }
    List<Payment> paymentList = new ArrayList<Payment>();

    try {
        VTDGen vg = new VTDGen();
        VTDGen vgHen = new VTDGen();
        boolean headercheck = true;
        if (vg.parseFile("src/main/resources/input/partitioner/" + resource.getFilename(), false)) {

            VTDNav vn = vg.getNav();
            AutoPilot ap = new AutoPilot(vn);
            ap.selectXPath("/root/Payment");
            // flb contains all the offset and length of the segments to be skipped
            FastLongBuffer flb = new FastLongBuffer(4);
            int i;
            byte[] xml = vn.getXML().getBytes();
            while ((i = ap.evalXPath()) != -1) {
                flb.append(vn.getElementFragment());
            }
            int size = flb.size();
            log.info("Payment Size {}", size);
            if (size != 0) {
                for (int k = 0; k < size; k++) {
                    String message = new String(xml, flb.lower32At(k), flb.upper32At(k), StandardCharsets.UTF_8);

                    ObjectMapper objectMapper = new ObjectMapper();
                    objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

                    Payment payment = objectMapper
                            .readValue(message, Payment.class);
                    paymentList.add(pcPayment);
                    index = pcPaymentList.size() + 1;

                }
            }
            log.info("Payment List:: {}", paymentList.size());
            log.info("Index::{}", index);
            return index > paymentList .size() ? null : (T) paymentList;
        }
    } catch (Exception e) {
        e.printStackTrace();
    }

    return null;
}

}  

SpringBatch 配置类

private final Logger logger = LoggerFactory.getLogger(SpringBatchConfig.class);

@Autowired
private JobBuilderFactory jobBuilderFactory;

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Autowired
ResourcePatternResolver resoursePatternResolver;


@Bean
public Job job() {
    return jobBuilderFactory.get("job").start(readpayment()).build();
}

@Bean
public JobLauncher jobLauncher() throws Exception {
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    jobLauncher.setJobRepository(jobRepository());
    jobLauncher.afterPropertiesSet();
    return jobLauncher;
}

@Bean
public JobRepository jobRepository() throws Exception {
    MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean();
    factory.setTransactionManager(new ResourcelessTransactionManager());
    return (JobRepository) factory.getObject();
}

@Bean
protected Step readpayment() {
    return stepBuilderFactory.get("readpayment").partitioner("paymentStep", partitioner(null))
            .step(paymentStep()).taskExecutor(taskExecutor()).build();
}

@Bean
protected Step paymentStep() {
    return stepBuilderFactory.get("paymentStep")
            .<Payment,Payment>chunk(10)
         .reader(xmlFileItemReader(null))
        .writer(writer()).build();
}


@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setMaxPoolSize(10);
    taskExecutor.setCorePoolSize(10);
    taskExecutor.setQueueCapacity(10);
    taskExecutor.afterPropertiesSet();
    return taskExecutor;
}

@Bean
@StepScope
ItemReader<Payment> xmlFileItemReader(@Value("#{stepExecutionContext[fileName]}") String filename) {
    VtdWholeItemReader<Payment> xmlFileReader = new VtdWholeItemReader<>();
    xmlFileReader.setResource(new ClassPathResource("input/partitioner/" + filename));
    return xmlFileReader;
}

@Bean
@StepScope
public CustomMultiResourcePartitioner partitioner(@Value("#{jobParameters['fileName']}") String fileName) {
    logger.info("fileName {}", fileName);
    CustomMultiResourcePartitioner partitioner = new CustomMultiResourcePartitioner();
    Resource[] resources;
    try {
        resources = resoursePatternResolver.getResources("file:src/main/resources/input/partitioner/*.xml");
    } catch (IOException e) {
        throw new RuntimeException("I/O problems when resolving the input file pattern.", e);
    }
    partitioner.setResources(resources);
    return partitioner;
}

@Bean
public ItemWriter<Payment> writer() {
    return new PaymentItemWriter();
}

PaymentItemWriter

@Override
public void write(List<? extends List<Payment>> items) throws Exception {

    log.info("Items {}", items.size());
}

Answer 1

可以尝试在 Spring 批处理配置类中将 ItemWriter bean 作为步骤范围 ["@StepScope"]