停止文件适配器

停止文件适配器

关注
发信
关注(28)粉丝(399)

来自控制器的 Spring Batch 集成启动/停止文件适配器

本文介绍了来自控制器的 Spring Batch 集成启动/停止文件适配器的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我不确定我的端点是否设置正确.我的最终目标是当我在两个适配器上调用 stop 时，如果将文件放置在入站通道适配器轮询的任一目录中，则不会处理该文件.事实并非如此——文件仍在被轮询和处理.

I am not sure if my endpoints are setup correctly. My end goal is when I call stop on both adapters, that if a file is placed in either directory where the inbound channel adapters are polling, the file will not be processed. Which is not the case -- files are still getting polled and processed.

目前，当代码运行时，从一开始就启动了 6 个任务计划程序:

Currently, when the code runs, from the very start, 6 Task Schedulers are launched:

2015-08-03 18:12:40,011 [task-scheduler-6] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:12:40,011 [task-scheduler-5] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:12:40,011 [task-scheduler-8] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:12:40,011 [task-scheduler-1] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:12:40,011 [task-scheduler-3] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:12:40,011 [task-scheduler-1] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'

当我发出停止令

@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.stop()

在其中一个适配器上，我剩下 5 个任务计划程序.这是发出的停止命令:

on one of the adapters, I am left with 5 Task Schedulers. Here is the stop command being issued:

2015-08-03 18:15:28,392 [http-8080-1] DEBUG org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'control'
2015-08-03 18:15:28,392 [http-8080-1] DEBUG org.springframework.integration.channel.DirectChannel - preSend on channel 'control', message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.stop(), headers={timestamp=1438650928392, id=8107d99e-667c-1e78-31a0-dbcac2a6b03e}]
2015-08-03 18:15:28,392 [http-8080-1] DEBUG org.springframework.integration.handler.ServiceActivatingHandler - ServiceActivator for [org.springframework.integration.handler.ExpressionCommandMessageProcessor@f275f321] received message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.stop(), headers={timestamp=1438650928392, id=8107d99e-667c-1e78-31a0-dbcac2a6b03e}]
2015-08-03 18:15:28,407 [http-8080-1] DEBUG org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'
2015-08-03 18:15:28,407 [http-8080-1] INFO  org.springframework.integration.endpoint.SourcePollingChannelAdapter - stopped mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter
2015-08-03 18:15:28,407 [http-8080-1] DEBUG org.springframework.integration.handler.ServiceActivatingHandler - handler 'ServiceActivator for [org.springframework.integration.handler.ExpressionCommandMessageProcessor@f275f321]' produced no reply for request Message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.stop(), headers={timestamp=1438650928392, id=8107d99e-667c-1e78-31a0-dbcac2a6b03e}]
2015-08-03 18:15:28,407 [http-8080-1] DEBUG org.springframework.integration.channel.DirectChannel - postSend (sent=true) on channel 'control', message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.stop(), headers={timestamp=1438650928392, id=8107d99e-667c-1e78-31a0-dbcac2a6b03e}]

这是第一次停止命令后生成的任务调度程序，现在只有 5 个:

And here are the resulting Task Schedulers after the first stop command, there are only 5 now:

2015-08-03 18:15:30,017 [task-scheduler-7] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:15:30,017 [task-scheduler-1] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:15:30,017 [task-scheduler-6] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:15:30,017 [task-scheduler-10] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:15:30,017 [task-scheduler-2] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'

当我在第二个适配器上发出第二个停止

When I issue the second stop on the second adapter

@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.stop()

命令如下:

2015-08-03 18:21:49,652 [http-8080-1] DEBUG org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'control'
2015-08-03 18:21:49,668 [http-8080-1] DEBUG org.springframework.integration.channel.DirectChannel - preSend on channel 'control', message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.stop(), headers={timestamp=1438651309652, id=d4ba5ca8-db6f-4ee9-564e-9a2df2f6fb4d}]
2015-08-03 18:21:49,668 [http-8080-1] DEBUG org.springframework.integration.handler.ServiceActivatingHandler - ServiceActivator for [org.springframework.integration.handler.ExpressionCommandMessageProcessor@f275f321] received message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.stop(), headers={timestamp=1438651309652, id=d4ba5ca8-db6f-4ee9-564e-9a2df2f6fb4d}]
2015-08-03 18:21:49,668 [http-8080-1] DEBUG org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'
2015-08-03 18:21:49,668 [http-8080-1] INFO  org.springframework.integration.endpoint.SourcePollingChannelAdapter - stopped mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter
2015-08-03 18:21:49,668 [http-8080-1] DEBUG org.springframework.integration.handler.ServiceActivatingHandler - handler 'ServiceActivator for [org.springframework.integration.handler.ExpressionCommandMessageProcessor@f275f321]' produced no reply for request Message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.stop(), headers={timestamp=1438651309652, id=d4ba5ca8-db6f-4ee9-564e-9a2df2f6fb4d}]
2015-08-03 18:21:49,668 [http-8080-1] DEBUG org.springframework.integration.channel.DirectChannel - postSend (sent=true) on channel 'control', message: GenericMessage [payload=@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.stop(), headers={timestamp=1438651309652, id=d4ba5ca8-db6f-4ee9-564e-9a2df2f6fb4d}]

我剩下 4 个任务计划程序:

I am left with 4 Task Schedulers:

2015-08-03 18:23:40,014 [task-scheduler-2] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:23:40,014 [task-scheduler-6] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:23:40,014 [task-scheduler-5] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 18:23:40,014 [task-scheduler-7] DEBUG org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'

这是 MecFilePoller 类的代码:

Here is the code MecFilePoller class:

@Configuration
@PropertySource("classpath:mec.properties")
@EnableIntegration
@IntegrationComponentScan
public class MecFilePoller {

private Job job;

private String fileParameterName;

@Autowired
MecProperty mecProperty;

@Bean
MessageChannel control() {
    return new DirectChannel();
}

@Bean
@ServiceActivator(inputChannel="control")
public ExpressionControlBusFactoryBean controlBus() {
    return new ExpressionControlBusFactoryBean();
}

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

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


@Bean
@InboundChannelAdapter(value = "ssnInboundFileChannel", poller = @Poller(cron="${mec/SSN_POLLER}"))
public MessageSource<File> fileMessageSourceSSN() {
     System.out.println("SSN POLLING...");
     FileReadingMessageSource source = initialSetUp();
     source.setDirectory(new File(mecProperty.getProperty(MecConstants.SSN_SFTP_WORKING_DIR)));
     System.out.println("enter fileMessageSource in following dir....." + mecProperty.getProperty(MecConstants.SSN_SFTP_WORKING_DIR));
     return source;
}

@Bean
@InboundChannelAdapter(value = "mdwInboundFileChannel", poller = @Poller(cron="${mec/MDW_POLLER}"))
public MessageSource<File> fileMessageSourceMDW() {
    System.out.println("MDW POLLING...");
    FileReadingMessageSource source = initialSetUp();
    source.setDirectory(new File(mecProperty.getProperty(MecConstants.MDW_SFTP_WORKING_DIR)));
    System.out.println("enter fileMessageSource in following dir....." + mecProperty.getProperty(MecConstants.MDW_SFTP_WORKING_DIR));
    return source;
}

private FileReadingMessageSource initialSetUp() {
    FileReadingMessageSource source = new FileReadingMessageSource();
    CompositeFileListFilter<File> compositeFileListFilter = new CompositeFileListFilter<File>();
    SimplePatternFileListFilter simplePatternFileListFilter = new SimplePatternFileListFilter("*.done");
    AcceptOnceFileListFilter<File> acceptOnceFileListFilter = new AcceptOnceFileListFilter<File>();

    compositeFileListFilter.addFilter(simplePatternFileListFilter);
    compositeFileListFilter.addFilter(acceptOnceFileListFilter);

    source.setFilter(compositeFileListFilter);
    return source;
}

@Transformer(inputChannel="mdwInboundFileChannel",outputChannel="mdwOutboundJobRequestChannel")
public JobLaunchRequest toRequestMDW(Message<File> message) {
    JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();
    jobParametersBuilder.addString(INPUT_FILE_NAME, message.getPayload().getAbsolutePath());
    return new JobLaunchRequest((Job) appContext.getBean("mecmdwJob"), jobParametersBuilder.toJobParameters());
}

@Transformer(inputChannel="ssnInboundFileChannel",outputChannel="ssnOutboundJobRequestChannel")
public JobLaunchRequest toRequestSSN(Message<File> message) {
    JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();
    jobParametersBuilder.addString(INPUT_FILE_NAME, message.getPayload().getAbsolutePath());
    return new JobLaunchRequest((Job) appContext.getBean("mecssnJob"), jobParametersBuilder.toJobParameters());
}

public void commandTypePollerRequest(String command, String type) {
    if(command.equals("stop") && type.equals("mdw")){
        control().send(new GenericMessage<String>(
                "@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.stop()"));
    }
    else if(command.equals("stop") && type.equals("ssn")){
        control().send(new GenericMessage<String>(
                "@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.stop()"));
    }
    else if(command.equals("start") && type.equals("mdw")){
        control().send(new GenericMessage<String>(
                "@'mecFilePoller.fileMessageSourceMDW.inboundChannelAdapter'.start()"));
    }
    else if(command.equals("start") && type.equals("ssn")){
        control().send(new GenericMessage<String>(
                "@'mecFilePoller.fileMessageSourceSSN.inboundChannelAdapter'.start()"));
    }
}

}

这是 MecPollerController 类:

Here is the MecPollerController class:

@Controller
public class MECPollerController {

@Autowired
MecFilePoller mecFilePoller;

@Autowired
private ApplicationContext appContext;

@Autowired
MessageChannel ssnOutboundJobRequestChannel;

@Autowired
MessageChannel mdwOutboundJobRequestChannel;

@RequestMapping(value = "ui/stopMdwPoller.action", method = RequestMethod.GET)
public void stopMdwPollerRequest() {
    mecFilePoller.commandTypePollerRequest("stop","mdw");
}

@RequestMapping(value = "ui/stopSsnPoller.action", method = RequestMethod.GET)
public void stopSsnPollerRequest() {
    mecFilePoller.commandTypePollerRequest("stop","ssn");
}

@RequestMapping(value = "ui/startMdwPoller.action", method = RequestMethod.GET)
public void startMdwPollerRequest() {
    mecFilePoller.commandTypePollerRequest("start","mdw");
}

@RequestMapping(value = "ui/startSsnPoller.action", method = RequestMethod.GET)
public void startSsnPollerRequest() {
    mecFilePoller.commandTypePollerRequest("start","ssn");
}

}

这是一个 Spring Batch 作业 job-mecmdw.xml 的 XML 配置:

Here is the XML config for the one Spring Batch job, job-mecmdw.xml:

<int:channel id="mdwInboundFileChannel" />
<int:channel id="mdwOutboundJobRequestChannel" />
<int:channel id="mdwJobLaunchReplyChannel" />
<int:annotation-config />
<bean class="org.springframework.integration.core.MessagingTemplate"></bean>

<!--    THIS WAS MOVED TO MecFilePoller
<int:transformer input-channel="mdwInboundFileChannel" output-channel="mdwOutboundJobRequestChannel">
    <bean
        class="org.batch.poller.MecFilePoller">
        <property name="job" ref="mecmdwJob" />
        <property name="fileParameterName" value="input.file.name" />
    </bean>
</int:transformer>
-->
<batch-int:job-launching-gateway request-channel="mdwOutboundJobRequestChannel" reply-channel="mdwJobLaunchReplyChannel" />
<int:logging-channel-adapter channel="mdwJobLaunchReplyChannel" />

这是另一个 Spring Batch 作业 job-mecssn.xml 的 XML 配置:

Here is the XML config for the other Spring Batch job, job-mecssn.xml:

<int:annotation-config />
<int:channel id="ssnInboundFileChannel" />
<int:channel id="ssnOutboundJobRequestChannel" />
<int:channel id="ssnJobLaunchReplyChannel" />

<!--    THIS WAS MOVED TO MecFilePoller
<int:transformer input-channel="ssnInboundFileChannel"
    output-channel="ssnOutboundJobRequestChannel">
    <bean
        class="org.batch.poller.MecFilePoller">
        <property name="job" ref="mecssnJob" />
        <property name="fileParameterName" value="input.file.name" />
    </bean>
</int:transformer>
-->

<batch-int:job-launching-gateway request-channel="ssnOutboundJobRequestChannel" reply-channel="ssnJobLaunchReplyChannel" />
<int:logging-channel-adapter channel="ssnJobLaunchReplyChannel" />

推荐答案

入站通道适配器由(至少两个)bean 组成；在这种情况下，一个 MessageSource 和一个 SourcePollingChannelAdapter.

Inbound channel adapters consist of (at least two) beans; in this case, a MessageSource and a SourcePollingChannelAdapter.

在您的示例中，消息源获取指定的 bean 名称 (filePoller).通道适配器的名称已生成，并将是

In your example, the message source gets the specified bean name (filePoller). The channel adapter's name is generated, and will be

filePoller.fileMessageSource.inboundChannelAdapter

(@Configuration 类 bean 名称 + 方法名称 + 类型)

(@Configuration class bean name + method name + type)

我们有一个开放 JIRA 问题来记录这些生成的 bean 名称.

We have an open JIRA issue to document these generated bean names.

由于名称是虚线，所以在使用控制总线时需要引用它:

Since the name is dotted, you'll need to quote it when using the control bus:

@'filePoller.fileMessageSource.inboundChannelAdapter'.stop()

编辑

我刚刚测试了这个，它工作得很好......

I just tested this and it works just fine...

@Configuration
@EnableIntegration
public class Foo {

    @Bean
    QueueChannel inboundFileChannel() {
        return new QueueChannel();
    }

    @Bean
    @InboundChannelAdapter(value = "inboundFileChannel", poller = @Poller(fixedDelay="1000"))
    public MessageSource<?> fileMessageSource() {
        MessageSource<?> source = mock(MessageSource.class);
        return source;
    }

    @Bean
    MessageChannel control() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel="control")
    public ExpressionControlBusFactoryBean controlBus() {
        return new ExpressionControlBusFactoryBean();
    }

    public static void main(String[] args) throws Exception {
        AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(Foo.class);
        Thread.sleep(2000);
        ctx.getBean("control", MessageChannel.class).send(
                new GenericMessage<String>("@'foo.fileMessageSource.inboundChannelAdapter'.stop()"));
        assertFalse(ctx.getBean(SourcePollingChannelAdapter.class).isRunning());
        Thread.sleep(2000);
        ctx.close();
    }
}

EDIT2:

我在上面的例子中睡了一些，看到这个......

I put some sleeps in my example above and see this...

2015-08-03 12:23:34,528 [task-scheduler-1] DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 12:23:35,530 [task-scheduler-2] DEBUG: org.springframework.integration.endpoint.SourcePollingChannelAdapter - Received no Message during the poll, returning 'false'
2015-08-03 12:23:35,615 [main] DEBUG: org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'control'
2015-08-03 12:23:35,616 [main] DEBUG: org.springframework.integration.channel.DirectChannel - preSend on channel 'control', message: GenericMessage [payload=@'foo.fileMessageSource.inboundChannelAdapter'.stop(), headers={id=d0049777-025c-5439-6ee7-d66bce34868a, timestamp=1438619015616}]
2015-08-03 12:23:35,616 [main] DEBUG: org.springframework.integration.handler.ServiceActivatingHandler - ServiceActivator for [org.springframework.integration.handler.ExpressionCommandMessageProcessor@1e0b4072] received message: GenericMessage [payload=@'foo.fileMessageSource.inboundChannelAdapter'.stop(), headers={id=d0049777-025c-5439-6ee7-d66bce34868a, timestamp=1438619015616}]
2015-08-03 12:23:35,617 [main] DEBUG: org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'foo.fileMessageSource.inboundChannelAdapter'
2015-08-03 12:23:35,618 [main] INFO : org.springframework.integration.endpoint.SourcePollingChannelAdapter - stopped foo.fileMessageSource.inboundChannelAdapter
2015-08-03 12:23:35,618 [main] DEBUG: org.springframework.integration.handler.ServiceActivatingHandler - handler 'ServiceActivator for [org.springframework.integration.handler.ExpressionCommandMessageProcessor@1e0b4072]' produced no reply for request Message: GenericMessage [payload=@'foo.fileMessageSource.inboundChannelAdapter'.stop(), headers={id=d0049777-025c-5439-6ee7-d66bce34868a, timestamp=1438619015616}]
2015-08-03 12:23:35,618 [main] DEBUG: org.springframework.integration.channel.DirectChannel - postSend (sent=true) on channel 'control', message: GenericMessage [payload=@'foo.fileMessageSource.inboundChannelAdapter'.stop(), headers={id=d0049777-025c-5439-6ee7-d66bce34868a, timestamp=1438619015616}]
2015-08-03 12:23:35,618 [main] DEBUG: org.springframework.beans.factory.support.DefaultListableBeanFactory - Returning cached instance of singleton bean 'foo.fileMessageSource.inboundChannelAdapter'

(停止后没有投票).

这篇关于来自控制器的 Spring Batch 集成启动/停止文件适配器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！

08-28 02:14
Powered by LMLPHP ©2024 停止文件适配器 0.035084