Spring 集成 FileReadingMessageSource 使用 UseWatchService

问题描述 投票：0回答：2

我使用 

FileReadingMessageSource
作为 Spring Integration ESB 中消息的来源。

我特别想使用

WatchServiceDirectoryScanner
。我在 Spring Boot 中使用基于注释的配置。

如果我不使用

WatchServiceDirectoryScanner
一切都很好，但是当我设置
setUseWatchService(true)
时我无法让它工作。

只有应用程序启动时目录中存在的文件才会在通道中生成新消息。在目录中复制或创建的文件不会生成任何消息。

我想使用 

WatchServiceDirectoryScanner
，因为我只想生成一条消息

这是我配置通道、适配器和源的代码：

    @Bean(name="source")
    public FileReadingMessageSource getFileMessageSource(){
        FileReadingMessageSource lm = new FileReadingMessageSource();
        lm.setBeanName("fileMessageSource");
        lm.setDirectory(new File("C:/DestDir"));
        lm.setAutoCreateDirectory(true);

        lm.setFilter(new AcceptAllFileListFilter<>());
        lm.setUseWatchService(true);
        lm.setWatchEvents(WatchEventType.CREATE);
        return lm;
    }

@Bean(name="channel")
    public PublishSubscribeChannel getFileChannel(){
        PublishSubscribeChannel psc = new PublishSubscribeChannel();
        psc.setLoggingEnabled(true);
        psc.setComponentName("channelexp");
        return psc;

    }


    @Bean
    @DependsOn({"source","channel"})
    public SourcePollingChannelAdapter  getChannelAdapter(){
        SourcePollingChannelAdapter spca = new SourcePollingChannelAdapter();
        FileReadingMessageSource frms = context.getBean(FileReadingMessageSource.class);
        PublishSubscribeChannel psc = context.getBean("channel",PublishSubscribeChannel.class);
        spca.setSource(frms);
        spca.setOutputChannel(psc);
        return spca;
    }

即使我使用

@InboundChannelAdapter
也没有什么区别

@Bean(name="source")
    @InboundChannelAdapter(value = "channel", poller = @Poller(fixedDelay = "1000"))
    public FileReadingMessageSource getFileMessageSource(){
        FileReadingMessageSource lm = new FileReadingMessageSource();
        lm.setDirectory(new File("C:/fromDir/"));
        lm.setFilter(new AcceptAllFileListFilter<>());
        lm.setUseWatchService(true);
        lm.setWatchEvents(WatchEventType.CREATE);
        return lm;
    }

我哪里做错了？

java spring spring-boot spring-integration
2个回答
3
投票

我在您的配置中没有看到轮询代码，但我想知道您是否还有其他内容：

  1. 你必须使用
    @EnableIntegration
  2. SourcePollingChannelAdapter
    应与 
    PollerMetadata
    3. 一起提供
  3. 或者只是考虑使用
    @InboundChannelAdapter
    （参见参考手册）
  4. 在 bean 定义过程中使用 
    context.getBean()
    看起来确实很奇怪。你应该在那里使用injection。前者用于运行时而不是初始化阶段。

编辑

Spring Integration Java DSL 示例：

@SpringBootApplication
public class FileChangeLineSeparator {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(FileChangeLineSeparator.class)
                .web(false)
                .run(args);
        System.out.println("Put a windows file with a .txt extension in /tmp/in,"
                + "\nthe file will be converted to Un*x and placed in"
                + "\n/tmp/out"
                + "\n\nHit enter to terminate");
        System.in.read();
        context.close();
    }

@Bean
@InboundChannelAdapter(value = "channel", poller = @Poller(fixedDelay = "1000"))
public FileReadingMessageSource getFileMessageSource() {
    FileReadingMessageSource lm = new FileReadingMessageSource();
    lm.setDirectory(new File("/tmp/in"));
    lm.setFilter(new AcceptAllFileListFilter<>());
    lm.setUseWatchService(true);
    lm.setWatchEvents(FileReadingMessageSource.WatchEventType.CREATE);
    return lm;
}

@Bean
public IntegrationFlow fileToFile() {
    return IntegrationFlows.from("channel")                 
                .transform(Transformers.fileToString())
                .transform("payload.replaceAll('\r\n', '\n')")
                .handle(Files.outboundAdapter("'/tmp/out'")
                        .autoCreateDirectory(true))
                .get();
    }

}

我最终创建了新文件，并且它们被正确拾取并处理。

1
投票

即使@Artem Bilan 的答案非常有用（这就是我接受它的原因），我也会发布自己的答案来展示我的最终解决方案。

这是完整的示例

我必须监视目录并在每次修改文件时创建一条消息。 我使用 Spring ESB 

FileReadingMessageSource
。 此外，我必须在运行时实例化所有组件，因此我使用了 Spring Integration Java DSL 
IntegrationFlowContext
。

这就是我在运行时启动源的方式（消息被发送到

"output"
通道）。

    @Autowired
    private IntegrationFlowContext intContext;

public void initDispatcher(String name,Long pollingFreq) throws HandlerConfigurationException {
        logger.info("Source "+ name +" Inizializing");


        IntegrationFlow sourceModifiedFlow = IntegrationFlows
                .from(this.getFileModifiedMessageSource(), s -> s.poller(Pollers.fixedRate(pollingFreq, TimeUnit.MILLISECONDS)))
                .channel("outputChannel").get();
        intContext.registration(sourceModifiedFlow).id(name)).autoStartup(true).register();
    }

这就是我创建实际的

FileReader
的方式。

 public FileReadingMessageSource getFileModifiedMessageSource() {
            FileReadingMessageSource lm = new FileReadingMessageSource();
            lm.setBeanName(String.format(SOURCE_MODIFIED_SUFFIX, name));
            lm.setDirectory(new File(readingDir));
            ModifiedOnlyFileScanner modifiedOnlyFileScanner = new ModifiedOnlyFileScanner(new File(readingDir));
            modifiedOnlyFileScanner.setFilter(new AcceptAllFileListFilter<>());
            lm.setScanner(modifiedOnlyFileScanner);

            return lm;
        }

默认的文件扫描仪（

org.springframework.integration.file.FileReadingMessageSource.WatchServiceDirectoryScanner
）对我来说不太好。主要有两个原因：

  1. 启动时为目录中存在的每个文件生成一条消息
  2. 即使我使用 
    fileReadSource.setWatchEvents(WatchEventType.MODIFY);
    配置阅读器，当在目录中创建新文件时，源也会发送一条消息。

所以我最终编写了自己的扫描仪。这是完整的代码：

public class ModifiedOnlyFileScanner extends DefaultDirectoryScanner implements Lifecycle {
    private final static Logger logger = LogManager.getLogger(FileDispatcher.class);


    private final ConcurrentMap<Path, WatchKey> pathKeys = new ConcurrentHashMap<Path, WatchKey>();

    private WatchService watcher;

    private WatchEvent.Kind<?>[] kinds;

    private File directory;

    @Override
    public void start() {
        try {
            this.watcher = FileSystems.getDefault().newWatchService();
        }
        catch (IOException e) {
            logger.error("Failed to create watcher for " + directory, e);
        }

        this.kinds = new WatchEvent.Kind<?>[]{StandardWatchEventKinds.ENTRY_MODIFY};
        walkDirectory(directory.toPath(), null);

    }

    public ModifiedOnlyFileScanner(File directory) {
        super();
        this.directory = directory;
    }

    @Override
    public void stop() {
        try {
            this.watcher.close();
            this.watcher = null;
            this.pathKeys.clear();
        }
        catch (IOException e) {
            logger.error("Failed to close watcher for " + directory, e);
        }
    }

    @Override
    public boolean isRunning() {
        return true;
    }

    @Override
    protected File[] listEligibleFiles(File directory) {
        Assert.state(this.watcher != null, "The WatchService has'nt been started");
        Collection<File> files = filesFromEvents();
        if(files.size()>0){
            logger.info("Detected Modified files "+files);
        }
        return files.toArray(new File[files.size()]);
    }

    private Set<File> filesFromEvents() {
        WatchKey key = this.watcher.poll();
        Set<File> files = new LinkedHashSet<File>();
        while (key != null) {
            File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile();
            for (WatchEvent<?> event : key.pollEvents()) {
                if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY ) {
                    Path item = (Path) event.context();
                    File file = new File(parentDir, item.toFile().getName());
                    if (file.exists()) {
                        //I do not dig into directories
                        if (!file.isDirectory()) {
                            files.remove(file);
                            files.add(file);
                        }
                    }
                    else {
                            logger.warn("A file [" + file + "] for the event [" + event.kind() +
                                    "] doesn't exist. Ignored.");
                    }
                }
            }
            key.reset();
            key = this.watcher.poll();
        }
        return files;
    }

    private Set<File> walkDirectory(Path directory, final WatchEvent.Kind<?> kind) {
        final Set<File> walkedFiles = new LinkedHashSet<File>();
        try {
            registerWatch(directory);
            Files.walkFileTree(directory, new SimpleFileVisitor<Path>() {

                @Override
                public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                    FileVisitResult fileVisitResult = super.preVisitDirectory(dir, attrs);
                    registerWatch(dir);
                    return fileVisitResult;
                }

                @Override
                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                    FileVisitResult fileVisitResult = super.visitFile(file, attrs);
                    if (!StandardWatchEventKinds.ENTRY_MODIFY.equals(kind)) {
                        walkedFiles.add(file.toFile());
                    }
                    return fileVisitResult;
                }

            });
        }
        catch (IOException e) {
            logger.error("Failed to walk directory: " + directory.toString(), e);
        }
        return walkedFiles;
    }

    private void registerWatch(Path dir) throws IOException {
        if (!this.pathKeys.containsKey(dir)) {
            if (logger.isDebugEnabled()) {
                logger.debug("registering: " + dir + " for file events");
            }
            WatchKey watchKey = dir.register(this.watcher, this.kinds);
            this.pathKeys.putIfAbsent(dir, watchKey);
        }
    }
}
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.