spring watchservicedirectoryscanner 的工作示例

Question

我正在努力实现WatchServiceDirectoryScanner。我想使用扫描仪监控新文件上传到目录+子目录。这将作为 Spring Boot MVC 微服务的一部分存在。我可以使用 Java 7 的 WatchService 来做到这一点，但更喜欢 spring 文件集成风格，AOP 风格

我在我的应用程序配置中将它注册为@Bean，但我正在努力弄清楚如何让它轮询和扫描一个目录，然后调用...某些东西（一个消息端点？）当一个文件被检测到。有没有人能够为我指出正确的方向，甚至在概念上如何做到这一点。我在任何地方都找不到这样的示例实现。

http://docs.spring.io/spring-integration/reference/html/files.html#_watchservicedirectoryscanner

http://docs.spring.io/spring-integration/api/org/springframework/integration/file/DefaultDirectoryScanner.html#listFiles-java.io.File-

这是我的 Spring appConfig：

public class appConfig {
    @Bean
    public DirectoryScanner scanner() {
        return new WatchServiceDirectoryScanner("/uploads/test");
    }
}

Answer 1

   # create one configuration file and bind an input file channel here

    <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:int="http://www.springframework.org/schema/integration"
        xmlns:int-file="http://www.springframework.org/schema/integration/file"
        xmlns:int-mail="http://www.springframework.org/schema/integration/mail"
        xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
        xsi:schemaLocation="http://www.springframework.org/schema/integration/mail http://www.springframework.org/schema/integration/mail/spring-integration-mail.xsd
            http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
            http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
            http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
            http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

        <int:annotation-config />
            <int:channel id="cfpFileIn"></int:channel>
    <int-file:inbound-channel-adapter id="cfpFileIn"
            directory="${cfp.flight.data.dir}" auto-startup="true" scanner="csvDirScanner">
            <int:poller fixed-delay="${cfp.flight.data.dir.polling.delay}"></int:poller>
        </int-file:inbound-channel-adapter>
    <bean id="csvDirScanner"
            class="org.springframework.integration.file.WatchServiceDirectoryScanner">
            <constructor-arg index="0" value="${cfp.flight.data.dir}" />
            <property name="filter" ref="csvCompositeFilter" />
            <property name="autoStartup" value="true" />
        </bean>

    <bean id="csvCompositeFilter"
            class="org.springframework.integration.file.filters.CompositeFileListFilter">
            <constructor-arg>
                <list>
                    <bean
                        class="org.springframework.integration.file.filters.SimplePatternFileListFilter">
                        <constructor-arg value="*.csv" />
                    </bean>
                    <ref bean="persistentFilter" />
                </list>
            </constructor-arg>
        </bean>

    <bean id="persistentFilter"
            class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter">
            <constructor-arg index="0" ref="metadataStore" />
            <constructor-arg index="1" name="prefix" value="" />
            <property name="flushOnUpdate" value="true" />
        </bean>

    <bean name="metadataStore"
            class="org.springframework.integration.metadata.PropertiesPersistingMetadataStore">
            <property name="baseDirectory" value="${metadata.dir}"></property>
        </bean>
</beans>

Answer 2

import java.io.File;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.stereotype.Component;
@Component
public class BatchJobScheduler  {

    private static final Logger logger = LoggerFactory.getLogger(BatchJobScheduler.class);
    @Autowired
protected JobLauncher jobLauncher;

    @Autowired
    @Qualifier(value = "job")
    private Job job;

    @ServiceActivator(inputChannel = "cfpFileIn")
    public void run(File file) {
        String fileName = file.getAbsolutePath();
        logger.info("BatchJobScheduler Running #################"+fileName);
        JobParameters jobParameters = new JobParametersBuilder().addString(
                "input.file", fileName).toJobParameters();


        try {

            JobExecution execution = jobLauncher.run(job,
                    jobParameters);
            logger.info(" BatchJobScheduler Exit Status : " + execution.getStatus() +"::"+execution.getAllFailureExceptions());

        } catch (JobExecutionAlreadyRunningException | JobRestartException
                | JobInstanceAlreadyCompleteException
                | JobParametersInvalidException e) {
            logger.error(" BatchJobScheduler Exit Status : Exception ::",e);
        }
    }
}