java - Spring Batch Processor 未运行 ItemProcessor Listener

Question

所以我在 Spring Batch 3.0.7.RELEASE 和 Spring 4.3.2.RELEASE 中遇到问题，监听器没有在我的 ItemProcessor< 中运行 类。 @StepScope 级别的常规注入(inject)适用于 @Value("#{jobExecutionContext['"+ Constants.SECURITY_TOKEN + "']}")，如下所示。但它不适用于 beforeProcess 或 beforeStep，我已经尝试了注释版本和接口(interface)版本。我几乎 100% 确定这在某个时候是有效的，但不明白为什么它停止了。

有什么想法吗？是不是我配置错了？

AppBatchConfiguration.java

@Configuration
@EnableBatchProcessing
@ComponentScan(basePackages = "our.org.base")
public class AppBatchConfiguration {

    private final static SimpleLogger LOGGER = SimpleLogger.getInstance(AppBatchConfiguration.class);

    private final static String OUTPUT_XML_FILE_PATH_PLACEHOLDER = null;
    private final static String INPUT_XML_FILE_PATH_PLACEHOLDER = null;

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    @Bean(name = "cimAppXmlReader")
    @StepScope
    public <T> ItemStreamReader<T> appXmlReader(@Value("#{jobParameters[inputXmlFilePath]}")
    String inputXmlFilePath) {
        LOGGER.info("Job Parameter => App XML File Path :" + inputXmlFilePath);
        StaxEventItemReader<T> reader = new StaxEventItemReader<T>();
        reader.setResource(new FileSystemResource(inputXmlFilePath));
        reader.setUnmarshaller(mecaUnMarshaller());
        reader.setFragmentRootElementNames(getAppRootElementNames());
        reader.setSaveState(false);

        // Make the StaxEventItemReader thread-safe
        SynchronizedItemStreamReader<T> synchronizedItemStreamReader = new SynchronizedItemStreamReader<T>();
        synchronizedItemStreamReader.setDelegate(reader);

        return synchronizedItemStreamReader;
    }

    @Bean
    @StepScope
    public ItemStreamReader<JAXBElement<AppIBTransactionHeaderType>> appXmlTransactionHeaderReader(@Value("#{jobParameters[inputXmlFilePath]}")
    String inputXmlFilePath) {
        LOGGER.info("Job Parameter => App XML File Path for Transaction Header :" + inputXmlFilePath);
        StaxEventItemReader<JAXBElement<AppIBTransactionHeaderType>> reader = new StaxEventItemReader<>();
        reader.setResource(new FileSystemResource(inputXmlFilePath));
        reader.setUnmarshaller(mecaUnMarshaller());

        String[] fragmentRootElementNames = new String[] {"AppIBTransactionHeader"};
        reader.setFragmentRootElementNames(fragmentRootElementNames);
        reader.setSaveState(false);

        return reader;
    }

    @Bean
    public Unmarshaller mecaUnMarshaller() {
        Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
        marshaller.setPackagesToScan(ObjectFactory.class.getPackage().getName());
        return marshaller;
    }

    @Bean
    public Marshaller uberMarshaller() {
        Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
        marshaller.setClassesToBeBound(ServiceRequestType.class);
        marshaller.setSupportJaxbElementClass(true);
        return marshaller;
    }

    @Bean(destroyMethod="") // To stop multiple close calls, see: http://stackoverflow.com/a/23089536
    @StepScope
    public ResourceAwareItemWriterItemStream<JAXBElement<ServiceRequestType>> writer(@Value("#{jobParameters[outputXmlFilePath]}")
    String outputXmlFilePath) {
        SyncStaxEventItemWriter<JAXBElement<ServiceRequestType>> writer = new SyncStaxEventItemWriter<JAXBElement<ServiceRequestType>>();

        writer.setResource(new FileSystemResource(outputXmlFilePath));
        writer.setMarshaller(uberMarshaller());
        writer.setSaveState(false);
        HashMap<String, String> rootElementAttribs = new HashMap<String, String>();
        rootElementAttribs.put("xmlns:ns1", "http://some.org/corporate/message/2010/1");
        writer.setRootElementAttributes(rootElementAttribs);
        writer.setRootTagName("ns1:SetOfServiceRequests");

        return writer;
    }

    @Bean
    @StepScope
    public <T> ItemProcessor<T, JAXBElement<ServiceRequestType>> appNotificationProcessor() {
        return new AppBatchNotificationItemProcessor<T>();
    }

    @Bean
    public ItemProcessor<JAXBElement<AppIBTransactionHeaderType>, Boolean> appBatchCreationProcessor() {
        return new AppBatchCreationItemProcessor();
    }


    public String[] getAppRootElementNames() {        
        //get list of App Transaction Element Names        
        return AppProcessorEnum.getValues();         
    }

    @Bean
    public Step AppStep() {
        // INPUT_XML_FILE_PATH_PLACEHOLDER and OUTPUT_XML_FILE_PATH_PLACEHOLDER will be overridden 
        // by injected jobParameters using late binding (StepScope)
        return stepBuilderFactory.get("AppStep")
                .<Object, JAXBElement<ServiceRequestType>> chunk(10)
                .reader(appXmlReader(INPUT_XML_FILE_PATH_PLACEHOLDER))
                .processor(appNotificationProcessor())
                .writer(writer(OUTPUT_XML_FILE_PATH_PLACEHOLDER))
                .taskExecutor(concurrentTaskExecutor())
                .throttleLimit(1)
                .build();

    }

    @Bean
    public Step BatchCreationStep() {
        return stepBuilderFactory.get("BatchCreationStep")
                .<JAXBElement<AppIBTransactionHeaderType>, Boolean>chunk(1)
                .reader(appXmlTransactionHeaderReader(INPUT_XML_FILE_PATH_PLACEHOLDER))
                .processor(appBatchCreationProcessor())
                .taskExecutor(concurrentTaskExecutor())
                .throttleLimit(1)
                .build();
    }

    @Bean
    public Job AppJob() {
        return jobBuilderFactory.get("AppJob")
                .incrementer(new RunIdIncrementer())
                .listener(AppJobCompletionNotificationListener())
                .flow(AppStep())
                .next(BatchCreationStep())
                .end()
                .build();
    }

    @Bean
    public JobCompletionNotificationListener AppJobCompletionNotificationListener() {
        return new JobCompletionNotificationListener();
    }

    @Bean
    public TaskExecutor concurrentTaskExecutor() {
        SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
        taskExecutor.setConcurrencyLimit(1);
        return taskExecutor;
    }
}

AppBatchNotificationItemProcessor.java

@StepScope
public class AppBatchNotificationItemProcessor<E> extends AppAbstractItemProcessor<E, JAXBElement<ServiceRequestType>> implements ItemProcessor<E, JAXBElement<ServiceRequestType>>, StepExecutionListener {

    // This is populated correctly
    @Value("#{jobExecutionContext['" + Constants.SECURITY_TOKEN + "']}")
    private SecurityToken securityToken;

    @Autowired
    private AppProcessorService processor;

    @Override
    public JAXBElement<ServiceRequestType> process(E item) throws BPException {
        // Do Stuff
        return srRequest;
    }

    @BeforeProcess
    public void beforeProcess(E item) {
        System.out.println("Doesn't execute");
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        // Doesn't execute
        System.out.println("Doesn't execute");
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        // Doesn't execute
        System.out.println("Doesn't execute");
    }

}

Answer 1

这是因为您在 @Bean 方法中返回接口(interface)而不是实现。恕我直言，在 Spring 中使用 java 配置时，您应该返回最具体的类型。原因如下:

通过 XML 配置时，您可以在 XML 配置中提供该类。这将实现公开给 Spring，以便可以发现并适当处理该类实现的任何接口(interface)。使用 java 配置时，@Bean 方法的返回类型将替代该信息。这就是问题所在。如果您的返回类型是一个接口(interface)，Spring 只知道该特定接口(interface)，而不知道实现可能实现的所有接口(interface)。通过尽可能返回具体类型，您可以让 Spring 深入了解您实际返回的内容，以便它可以更好地为您处理各种注册和连接用例。

对于您的具体示例，由于您返回的是 ItemProcessor 并且它是步骤范围的(因此是代理的)，因此 Spring 所知道的只是 ItemProcessor 所期望的方法/行为> 界面。如果您返回实现 (AppBatchNotificationItemProcessor)，则可以自动配置其他行为。