尝试将 Spring Batch 与 Spring 集成集成得到错误

Question

下面的类用于轮询目录中的文件，并在目录中接收到文件后触发 spring 批处理。我遇到了一些错误，我无法弄清楚。请指教。

另外，如果有一些示例代码示例可以做到这一点，请让我参考那个位置。

@Configuration
class FilePollingIntegrationFlow {

    @Autowired
    private ApplicationContext applicationContext;

     // this is the integration flow that foirst polls for messages and then trigger the spring batch job
    @Bean
    public IntegrationFlow inboundFileIntegration(@Value("${inbound.file.poller.fixed.delay}") long period,
                                                  @Value("${inbound.file.poller.max.messages.per.poll}") int maxMessagesPerPoll,
                                                  TaskExecutor taskExecutor,
                                                  MessageSource<File> fileReadingMessageSource,
                                                 JobLaunchingGateway jobLaunchingGateway) {



        return IntegrationFlows.from(fileReadingMessageSource,
                c -> c.poller(Pollers.fixedDelay(period)
                        .taskExecutor(taskExecutor)
                        .maxMessagesPerPoll(maxMessagesPerPoll)))
                  .transform(Transformers.fileToString())
                         .channel(ApplicationConfiguration.INBOUND_CHANNEL)                   
                  .handle((p, h) -> {
        System.out.println("Testing:::::"+p);
        return p;
 })
 .handle(fileMessageToJobRequest())
.handle(jobLaunchingGateway(),"toRequest")
 .channel(MessageChannels.queue())
                 .get();
    }

    @Bean
    public FileMessageToJobRequest fileMessageToJobRequest() {
        FileMessageToJobRequest fileMessageToJobRequest = new FileMessageToJobRequest();
        fileMessageToJobRequest.setFileParameterName("input.file.name");
      //  fileMessageToJobRequest.setJob(personJob());
        System.out.println("FilePollingIntegrationFlow::fileMessageToJobRequest::::Job launched successfully!!!");
        return fileMessageToJobRequest;
    }

    @Bean
    public JobLaunchingGateway jobLaunchingGateway() {
        SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
      //  simpleJobLauncher.setJobRepository(jobRepository);
        simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());
        JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
        System.out.println("FilePollingIntegrationFlow::jobLaunchingGateway::::Job launched successfully!!!");
        return jobLaunchingGateway;
    }

//This is another class used for batch job trigger

public class FileMessageToJobRequest {
    private Job job;
    private String fileParameterName;

    public void setFileParameterName(String fileParameterName) {
        this.fileParameterName = fileParameterName;
    }

    public void setJob(Job job) {
        this.job = job;
    }

    @Transformer
    public JobLaunchRequest toRequest(Message<File> message) {
        JobParametersBuilder jobParametersBuilder =
            new JobParametersBuilder();

        jobParametersBuilder.addString(fileParameterName,
            message.getPayload().getAbsolutePath());

        return new JobLaunchRequest(job, jobParametersBuilder.toJobParameters());
    }
}

我收到以下错误：

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'inboundFileIntegration' defined in class path resource [com/porterhead/integration/file/FilePollingIntegrationFlow.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.dsl.IntegrationFlow]: Factory method 'inboundFileIntegration' threw exception; nested exception is java.lang.IllegalArgumentException: Target object of type [class org.springframework.batch.integration.launch.JobLaunchingGateway] has no eligible methods for handling Messages.
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:599)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1123)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1018)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:510)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:482)
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:306)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:230)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:302)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:197)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:772)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:839)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:538)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:766)
    at org.springframework.boot.SpringApplication.createAndRefreshContext(SpringApplication.java:361)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:307)
    at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:134)
    at com.porterhead.Application.main(Application.java:23)
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.dsl.IntegrationFlow]: Factory method 'inboundFileIntegration' threw exception; nested exception is java.lang.IllegalArgumentException: Target object of type [class org.springframework.batch.integration.launch.JobLaunchingGateway] has no eligible methods for handling Messages.
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:189)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:588)
    ... 16 common frames omitted
Caused by: java.lang.IllegalArgumentException: Target object of type [class org.springframework.batch.integration.launch.JobLaunchingGateway] has no eligible methods for handling Messages.
    at org.springframework.integration.util.MessagingMethodInvokerHelper.findHandlerMethodsForTarget(MessagingMethodInvokerHelper.java:494)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:226)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:135)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:139)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.<init>(MethodInvokingMessageProcessor.java:52)
    at org.springframework.integration.handler.ServiceActivatingHandler.<init>(ServiceActivatingHandler.java:45)
    at org.springframework.integration.dsl.IntegrationFlowDefinition.handle(IntegrationFlowDefinition.java:982)
    at org.springframework.integration.dsl.IntegrationFlowDefinition.handle(IntegrationFlowDefinition.java:964)
    at com.porterhead.integration.file.FilePollingIntegrationFlow.inboundFileIntegration(FilePollingIntegrationFlow.java:85)
    at com.porterhead.integration.file.FilePollingIntegrationFlow$$EnhancerBySpringCGLIB$$c1cfa1e9.CGLIB$inboundFileIntegration$1(<generated>)
    at com.porterhead.integration.file.FilePollingIntegrationFlow$$EnhancerBySpringCGLIB$$c1cfa1e9$$FastClassBySpringCGLIB$$4ce6110e.invoke(<generated>)
    at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:228)
    at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:356)
    at com.porterhead.integration.file.FilePollingIntegrationFlow$$EnhancerBySpringCGLIB$$c1cfa1e9.inboundFileIntegration(<generated>)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:162)
    ... 17 common frames omitted

请指教。

Answer 1

由于您的FileMessageToJobRequest.toRequest() 标有@Transformer，您应该考虑改用.transform()。

我还看到您将 toRequest 方法名称用于 JobLaunchingGateway，这绝对是错误的。所以，正确的做法是这样的：

.transform(fileMessageToJobRequest())
.handle(jobLaunchingGateway())

您正在寻找的样本在 Spring Batch Reference Manual。