动态集合上的 Flux

Question

好吧，我有点困惑我应该如何将 Reactor 模式与 Spring 的 Webflux/Reactor API 一起使用。

假设我在不断添加文件的目录中。每当出现一个新文件时，我的应用程序就应该可以处理它。如果队列已满，则应忽略新文件（因此没有 FileWatcher），直到队列中有 space。

以下只是伪代码，我试图理解大概的思路，而不是具体的实现细节。

Class 目录观察员

管理一个计划任务，该任务每 2 秒检查一次是否出现新文件。如果是，我们尝试将它们添加到队列中。

@PostConstruct
public void initialize() {

    this.flux = Flux.generate(consumer -> {
        consumer.next(this.deque.pop()); 
    });

    this.flux.log();
}

@Scheduled(fixedRate = 2000)
public void checkDirectory() {
   // list files, for each add to deque (if not present)
}

public Flux<Path> getObserver() {
    return this.flux; 
}

Class 文件处理器

一个一个地消耗队列中的项目。

@PostConstruct
private void subscribe() {
    this.observeDirectorytask.getObserver().subscribe(path ->  {
        log.info("Processing '{}'", path.getFileName());

        // process file, delete it once done
    }); 
}

这种方法是否有意义？如果是，我需要做什么才能在新项目添加到队列时触发我的订阅（现在这只在启动时执行一次）。

更新

这是我的工作实现：

public class DirectoryObserverTask {

@Autowired
private Path observedDirectory;

private Consumer<Path> newFilesConsumer;
private Consumer<Throwable> errorsConsumer;

private Flux<Path> flux; 


@PostConstruct
public void init() {
    this.observedDirectory = Paths.get(importDirectoryProperty);
}

public void subscribe(Consumer<Path> consumer) {
    if(this.flux == null) {
        this.flux = Flux.push(sink -> {
            this.onError(err -> sink.error(err));
            this.onNewFile(file -> sink.next(file));
        }); 
        this.flux = this.flux.onBackpressureBuffer(10,  BufferOverflowStrategy.DROP_LATEST); 
    }

    this.flux.subscribe(consumer); 

}


@Scheduled(fixedRate = 2000)
public void checkDirectoryContent() throws IOException {    
    Files.newDirectoryStream(this.observedDirectory).forEach(path -> {
        this.newFilesConsumer.accept(path);
    });
}

public void onNewFile(Consumer<Path> newFilesConsumer) {
    this.newFilesConsumer = newFilesConsumer;
}

public void onError(Consumer<Throwable> errorsConsumer) {
    this.errorsConsumer = errorsConsumer;
}

}

和消费者

@Autowired
private DirectoryObserverTask observeDirectorytask;

@PostConstruct
private void init() {
    observeDirectorytask.subscribe(path -> {
        this.processPath(path);
    });
}

public void processPath(Path t) {
    Mono.justOrEmpty(t)
        .subscribe(path -> {
            // handle the file 
            path.toFile().delete();
        });
}

Answer 1

您不需要使用队列，此行为已内置。

我会这样做：

使用文件观察器检测更改。
将更改推送到 Flux<File>。
根据要求，限制排队事件的数量（使用背压）：

filesFlux.onBackPressureBuffer(10, BufferOverflowStrategy.DROP_LATEST)
照常订阅。

背压的错误解释是："what to do when we can't process elements fast enough"。

在这种情况下，我们最多缓冲 10 个元素，然后在缓冲区已满时删除最新的元素。

更新：创建Flux的方法有很多种。在这种特殊情况下，我会看一下 create 或 push 方法（参见 documentation）。

示例：假设您有一个 FileWatchService，您可以在其中注册检测到新文件时的回调和出现错误时的回调。你可以这样做：

FileWatchService watcher = ...

Flux<File> fileFlux = Flux.push(sink -> {
    watcher.onError(err -> sink.error(err));
    watcher.onNewFile(file -> sink.next(file));
});

fileFlux
    .onBackPressureBuffer(10, BufferOverflowStrategy.DROP_LATEST)
    .subscribe(...)

动态集合上的 Flux

Flux on a dynamic collection

java

spring

project-reactor

spring-webflux

更新