问题：

复制xml spring批处理应用程序中的头标记

仲俊豪

2023-03-14

我在＜code＞spring boot。Spring Boot版本为＜code＞2.3.3.RELEASE。

我打算实现的目标

我必须读取一个xml文件，其中包含数千个带有头标签（fileInformation）的Transaction。在事务上执行一些业务逻辑，然后使用事务中的更新值将文件写回。我使用StaxEventItemReader读取文件，使用StaxEventItemWriter写入文件。然后我有几个ItemProcers来处理业务逻辑。Xml文件看起来像：

<?xml version="1.0" encoding="UTF-8"?>
<reportFile>
   <fileInformation>
      <sender>200GH7XZ60</sender>
      <timestamp>2020-12-23T09:05:34Z</timestamp>
      <environment>PRO</environment>
      <version>001.60</version>
   </fileInformation>
   <record>
      <transaction>
         <buyer><buyer/>
      </transaction>
      <transaction>
         <buyer><buyer/>
      </transaction>
      <transaction>
         <buyer><buyer/>
      </transaction>
   </record>
</reportFile>

我面临的问题是标头标记的值。

我已经配置了OmegaXmlHeaderCallBack，它会生成所需的头标签，但这些标签中的值应该从输入文件中复制。据我所知，StaxWriterCallback是在读取器、处理器和写入器之前初始化的。所以我无法使用后期绑定注入值。这看起来像是一个基本要求，但在stackoverflow上找不到任何解决方案。

下面是用于配置 spring 批处理作业的代码段。

@Slf4j
@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

@Autowired
PIExtractorItemProcessor pIExtractorItemProcessor;

@Autowired
JobBuilderFactory jobBuilderFactory;
 
@Autowired
StepBuilderFactory stepBuilderFactory;

@Value( "${eugateway.batch.chunk.size}" )
private int chunkSize;

@Bean
public Step jobStep(ItemStreamReader<CustomHeaderTransactionXmlElement> reader,
        CompositeItemProcessor<CustomHeaderTransactionXmlElement, 
        ProcessorWriterDto> processor,
        CompositeItemWriter<ProcessorWriterDto> writer,
        EdsClientItemWriteListener<ProcessorWriterDto> writeListener, 
        StepBuilderFactory stepBuilderFactory) {
    return stepBuilderFactory.get("extractAndReplacePersonalDataStep")
            .<CustomHeaderTransactionXmlElement, ProcessorWriterDto>chunk(chunkSize)
            .reader(reader)
            .processor(processor)
            .listener(writeListener)
            .writer(writer)
            .build();
}

@Bean
public Job extractPersonalDataJob(Step jobStep, JobResultListener jobListener,
        JobBuilderFactory jobBuilderFactory) {
    return jobBuilderFactory.get("extractAndReplacePersonalDataJob")
            .incrementer(new RunIdIncrementer())
            .start(jobStep)
            .listener(jobListener)
            .build();
}

@Bean
@StepScope
public ItemStreamReader<CustomHeaderTransactionXmlElement> itemReader(@Value("#{jobParameters[file.path]}") String path) {
    Jaxb2Marshaller transactionMarshaller = new Jaxb2Marshaller();
    transactionMarshaller.setClassesToBeBound (FileInformation.class, TransactionPositionReport.class);
    log.info("Generating StaxEventItemReader");

    return new StaxEventItemReaderBuilder<CustomHeaderTransactionXmlElement>()
            .name("headerTransaction")
            .resource(new FileSystemResource(new FileSystemResource(path)))
            .addFragmentRootElements("fileInformation", "transaction")
            .unmarshaller(transactionMarshaller)
            .build();
}

@Bean
@StepScope
OmegaXmlHeaderCallBack getOmegaXmlHeaderCallBack(@Value("#{jobExecutionContext['header.sender']}") String sender,
        @Value("#{jobExecutionContext['header.timestamp']}") String timestamp,
        @Value("#{jobExecutionContext['header.environment']}") String environment,
        @Value("#{jobExecutionContext['header.version']}") String version){
    return new OmegaXmlHeaderCallBack(sender, timestamp, environment, version);
}

@Bean
@StepScope
OmegaXmlFooterCallBack getOmegaXmlFooterCallBack(){
    return new OmegaXmlFooterCallBack();
}

@StepScope
@Bean(name = "staxTransactionWriter")
public StaxEventItemWriter<TransactionPositionReport> staxTransactionItemWriter(@Value("#{jobExecutionContext['header.sender']}") String sender,
        @Value("#{jobExecutionContext['header.timestamp']}") String timestamp,
        @Value("#{jobExecutionContext['header.environment']}") String environment,
        @Value("#{jobExecutionContext['header.version']}") String version) {
    String exportFilePath = "C:\\Users\\sasharma\\Documents\\TO_BE_DELETED\\eugateway\\outputfile.xml";
    Resource exportFileResource = new FileSystemResource(exportFilePath);

    Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
    marshaller.setSupportDtd(true);
    marshaller.setSupportJaxbElementClass(true);
    marshaller.setClassesToBeBound(TransactionPositionReport.class);

    return new StaxEventItemWriterBuilder<TransactionPositionReport>()
            .name("transactionWriter")
            .version("1.0")
            .resource(exportFileResource)
            .marshaller(marshaller)
            .rootTagName("reportFile")
            .headerCallback(getOmegaXmlHeaderCallBack(sender, timestamp, environment, version))
            .footerCallback(getOmegaXmlFooterCallBack())
            .shouldDeleteIfEmpty(true)
            .build();
}

@Bean
@StepScope
public PIExtractorItemProcessor extractItemProcessor() {
    log.info("Generating PIExtractorItemProcessor");
    return new PIExtractorItemProcessor();
}

@Bean
public PIRemoverItemProcessor removeItemProcessor() {
    log.info("Generating PIRemoverItemProcessor");
    return new PIRemoverItemProcessor();
}

@Bean
@StepScope
CompositeItemProcessor<CustomHeaderTransactionXmlElement, ProcessorWriterDto> extractAndRemoveItemProcessor() {
    log.info("Generating CompositeItemProcessor");
    CompositeItemProcessor<CustomHeaderTransactionXmlElement, ProcessorWriterDto> itemProcessor = new CompositeItemProcessor<>();
    itemProcessor.setDelegates((List<? extends ItemProcessor<?, ?>>) Arrays.asList(extractItemProcessor(), removeItemProcessor()));
    return itemProcessor;
}

@Bean
@StepScope
public EdsClientItemWriter<ProcessorWriterDto> edsClientItemWriter() {
    log.info("Generating EdsClientItemWriter");
    return new EdsClientItemWriter<>();
}

@Bean
@StepScope
public OmegaXmlFileWriter<ProcessorWriterDto> omegaXmlFileWriter(@Value("#{jobExecutionContext['header.sender']}") String sender,
        @Value("#{jobExecutionContext['header.timestamp']}") String timestamp,
        @Value("#{jobExecutionContext['header.environment']}") String environment,
        @Value("#{jobExecutionContext['header.version']}") String version) {
    log.info("Generating OmegaXmlFileWriter");
    return new OmegaXmlFileWriter(staxTransactionItemWriter(sender, timestamp, environment, version));
}


@Bean
@StepScope
public CompositeItemWriter<ProcessorWriterDto> compositeItemWriter(@Value("#{jobExecutionContext['header.sender']}") String sender,
        @Value("#{jobExecutionContext['header.timestamp']}") String timestamp,
        @Value("#{jobExecutionContext['header.environment']}") String environment,
        @Value("#{jobExecutionContext['header.version']}") String version) {
    log.info("Generating CompositeItemWriter");
    CompositeItemWriter<ProcessorWriterDto> compositeItemWriter = new CompositeItemWriter<>();
    compositeItemWriter.setDelegates(Arrays.asList(edsClientItemWriter(), omegaXmlFileWriter(sender, timestamp, environment, version)));
    return compositeItemWriter;
 }  
}

下面是OmegaXmlHeaderCallBack类。由于没有后期绑定，我总是最终在标头标记中获得空值。

@Slf4j
public class OmegaXmlHeaderCallBack implements StaxWriterCallback {
    private String sender;
    private String timestamp;
    private String environment;
    private String version;
    
    public OmegaXmlHeaderCallBack(String sender, String timestamp, String environment, String version) {
        super();
        this.sender = sender;
        this.timestamp = timestamp;
        this.environment = environment;
        this.version = version;
    }

    @Override
    public void write(XMLEventWriter writer) {
        XMLEventFactory factory = XMLEventFactory.newInstance();
        try {
            writer.add(factory.createStartElement("", "", "fileInformation"));

            writer.add(factory.createStartElement("", "", "sender"));
            writer.add(factory.createCharacters(sender));
            writer.add(factory.createEndElement("", "", "sender"));


            writer.add(factory.createStartElement("", "", "timestamp"));
            writer.add(factory.createCharacters(timestamp));
            writer.add(factory.createEndElement("", "", "timestamp"));

            writer.add(factory.createStartElement("", "", "environment"));
            writer.add(factory.createCharacters(environment));
            writer.add(factory.createEndElement("", "", "environment"));

            writer.add(factory.createStartElement("", "", "version"));
            writer.add(factory.createCharacters(version));
            writer.add(factory.createEndElement("", "", "version"));
            
            writer.add(factory.createEndElement("", "", "fileInformation"));
            
            writer.add(factory.createStartElement("", "", "record"));
            
        } catch (XMLStreamException e) {
            log.error("Error writing OMEGA XML Header: {}", e.getMessage());
            throw new OmegaXmlHeaderWriterException(e.getMessage());
        }
    }
}

ItemProcessor 的代码如下所示。我正在将标头数据设置为执行上下文，该执行上下文旨在由headerCallback读取（可悲的是不会发生）。

@Slf4j
public class PIExtractorItemProcessor implements ItemProcessor<CustomHeaderTransactionXmlElement, ProcessorWriterDto> {

    @Autowired
    PersonalDataExtractor personalDataExtractor;
    
    @Value("#{jobParameters['submission.account']}") 
    private String subAccntId;
    
    @Value("#{stepExecution}")
    private StepExecution stepExecution;
    
    @Override
    public ProcessorWriterDto process(CustomHeaderTransactionXmlElement headerTransactionElement) throws Exception {
        FileInformation header = null;
        TransactionPositionReport transaction = null;
        if(headerTransactionElement instanceof FileInformation) {
            header = (FileInformation)headerTransactionElement;
            stepExecution.getExecutionContext().put("header.sender", header.getSender());
            stepExecution.getExecutionContext().put("header.timestamp", header.getTimestamp());
            stepExecution.getExecutionContext().put("header.environment", header.getEnvironment());
            stepExecution.getExecutionContext().put("header.version", header.getVersion());
            log.debug("Header {} found.", header.toString());
            return null;
        } else {
            transaction = (TransactionPositionReport)headerTransactionElement;
            log.debug("NO header info found for transaction {}", transaction.getProcessingDetails().getCustomerTransactionId());
            log.info("Extracting personal data for transaction customer id {} and create EDS requestDto.", transaction.getProcessingDetails().getCustomerTransactionId());
            ProcessorWriterDto transferObject = new ProcessorWriterDto();
            transferObject.setEdsRequestDtoList(personalDataExtractor.extract(transaction, subAccntId));
            transferObject.setTransaction(transaction);
            return transferObject;
        }
    }
}

我推荐的链接：

访问执行上下文值标头回调
如何通过在 ItemReader 中修改 setFragmentRootElementNames 和封送拆收器来读取标头标记

共有1个答案

袁晟

2023-03-14

你的步伐太大了。我会把事情分成两步:

步骤1：提取文件信息头并将其置于作业执行上下文中
步骤2：从执行上下文中读取文件信息头，并将其用于该步骤所需的任何步骤范围bean（例如，本例中的stax回调）

下面是一个简单的例子：

import java.io.IOException;
import java.io.Serializable;
import java.util.Map;

import javax.xml.stream.XMLEventWriter;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.xml.StaxWriterCallback;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableBatchProcessing
public class SO67909123 {

    @Bean
    public Step extractHeaderStep(StepBuilderFactory steps) {
        return steps.get("extractHeaderStep")
                .tasklet((contribution, chunkContext) -> {
                    Map<String, Object> jobParameters = chunkContext.getStepContext().getJobParameters();
                    String inputFile = (String) jobParameters.get("file");
                    FileInformation fileInformation = extractFileInformation(inputFile);
                    ExecutionContext jobExecutionContext =  chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext();
                    jobExecutionContext.put("file.information", fileInformation);
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private FileInformation extractFileInformation(String inputFile) {
        // TODO extract header from inputFile
        FileInformation fileInformation = new FileInformation();
        fileInformation.sender = "200GH7XKDGO3GLZ60";
        fileInformation.version = "001.60";
        return fileInformation;
    }

    @Bean
    public Step processFile(StepBuilderFactory steps) {
        return steps.get("processFile")
                .tasklet((contribution, chunkContext) -> { // Change this to a chunk-oriented tasklet
                    Map<String, Object> jobExecutionContext = chunkContext.getStepContext().getJobExecutionContext();
                    FileInformation fileInformation = (FileInformation) jobExecutionContext.get("file.information");
                    System.out.println("Step 2: " + fileInformation);
                    return RepeatStatus.FINISHED;
        }).build();
    }
    
    @Bean
    @StepScope
    public StaxWriterCallback staxWriterCallback(@Value("#{jobExecutionContext['file.information']}") FileInformation fileInformation) {
        return new StaxWriterCallback() {
            @Override
            public void write(XMLEventWriter writer) throws IOException {
                // use fileInformation as needed here 
            }
        };
    }

    @Bean
    public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
        return jobs.get("job")
                .start(extractHeaderStep(steps))
                .next(processFile(steps))
                .build();
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(SO67909123.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        JobParameters jobParameters = new JobParametersBuilder()
                .addString("file", "transactions.xml")
                .toJobParameters();
        jobLauncher.run(job, jobParameters);
    }

    static class FileInformation implements Serializable {
        private String sender;
        private String version;
        // other fields

        @Override
        public String toString() {
            return "FileInformation{sender='" + sender + '\'' + ", version='" + version + '\'' + '}';
        }
    }

}

这个例子展示了这个想法。您只需编写从文件中提取xml标记的代码段（只需标题，请参见TODO）。该示例中的＜code＞StaxWriterCallback＜/code＞是一个步骤范围的bean，可以使用执行上下文中的头。步骤2中的其他步骤范围组件也可以以相同的方式配置（处理器、侦听器等）。

类似资料：

Azure批处理应用程序洞察

在运行.NET Core3.1控制台应用程序时，我正在尝试获取应用程序的见解，以便在azure批处理作业/任务中工作。 https://docs.microsoft.com/en-us/Azure/Batch/Monitor-Application-Insights https://docs.microsoft.com/en-us/Azure/Azure-monitor/app/worker-se
设置所有处理程序的标头

问题内容：现在看起来像这样有没有办法一次将缓存标头设置给所有处理程序？问题答案：包裹多路复用器而不是单个处理程序。请注意，使用的处理程序时的处理程序参数是。另外，将和处理程序添加到。
整个应用程序的LibGDX公共sprite批处理

这里手头的问题是，在Game类中有一个公共SpriteBatch是不是一个好主意，然后所有屏幕都使用它。这将避免在活动屏幕发生变化时重新分配对象。然而，我看到人们在每一个屏幕上都使用新的和私有的SpriteBatchs。为什么人们要这样做？我是不是漏了什么？
在控制台应用程序中处理任务

要在控制台应用程序中开始使用Hangfire，您需要首先将Hangfire包安装到控制台应用程序。因此，使用您的软件包管理器控制台窗口进行安装： PM> Install-Package Hangfire.Core 然后添加任务存储安装所需的软件包。例如，使用SQL Server： PM> Install-Package Hangfire.SqlServer 仅需 Hangfire.Core 软件包
spring批处理程序链

我需要依次执行七个不同的流程（一个接一个）。数据存储在Mysql中。我正在考虑以下选项，如果我错了，或者有更好的解决方案，请纠正我。要求： > 需要分块处理数据。我的解决方案和问题：数据读取：使用JdbcCursorItemReader读取数据，因为这是性能最好的db阅读器-但是，SQL非常复杂，所以我可能不得不考虑使用JdbcTemboard的自定义ItemReader？这让我在处理数据时
向每个处理程序添加响应标头，而无需重复同一行

问题内容：我正在写一个小型网站，对于每个页面，我都会在其标题中放置一个服务器名称：我想知道是否有一种方法可以设置http.ResponseWriter的默认服务器名称，因此我不必一遍又一遍地使用同一行？问题答案：创建一个包装器以设置标题：包装单个处理程序或传递给ListenAndServe的根处理程序：

复制xml spring批处理应用程序中的头标记

共有1个答案

相关问答

相关文章

相关阅读

相关工具

相关文档