Reactor：创建 Monos 到 Flux 的函数

Question

基本上，我正在 Spring Boot 中制作队列处理器，并希望使用 Reactor 进行异步。我已经使一个函数需要永远循环，因为它是从队列中拉出然后将项目标记为已处理的函数。

这是有效的阻止版本订阅返回单声道

while(true) {
    manager.Subscribe().block()
}

我不知道如何把它变成一个 Flux 我已经查看了一个区间、生成、创建等，如果不调用 block() 就无法工作

这是我尝试过的一个示例

Flux.generate(() -> manager,
    (state, sink) -> {
        state.Subscribe().block();
        sink.next("done");
        return state;
    }));

作为 Reactor 的新手，我无法找到任何关于循环和同步处理 Monos 而不会阻塞的内容。

以下是使用 AWS Java SDK v2 的订阅方法的作用：

public Mono Subscribe() {
    return Mono.fromFuture(_client.receiveMessage(ReceiveMessageRequest.builder()
            .waitTimeSeconds(10)
            .queueUrl(_queueUrl)
            .build()))
            .filter(x -> x.messages() != null)
            .flatMap(x -> Mono.when(x.messages()
                    .stream()
                    .map(y -> {
                        _log.warn(y.body());
                        return Mono.fromFuture(_client.deleteMessage(DeleteMessageRequest.builder()
                                .queueUrl(_queueUrl)
                                .receiptHandle(y.receiptHandle())
                                .build()));
                    }).collect(Collectors.toList())));
}

基本上，我只是轮询一个 SQS 队列，删除消息然后我想再做一次。这对我来说只是探索性的。

谢谢！

Answer 1

您需要两件事：一种在循环中订阅的方法，以及一种确保在每次迭代中有效调用Subscribe() 方法的方法（因为需要重新创建Future）。

repeat() 是一个内置运算符，一旦源完成，它将重新订阅其源。如果源出错，则重复循环停止。最简单的变体继续这样做Long.MAX_VALUE 次。

唯一的问题是，在您的情况下，Subscribe() 中的Mono 必须在每次迭代时重新创建。

为此，您可以将 Subscribe() 调用包装在 defer 中：它会在每次发生新订阅时重新调用该方法，包括每次重复尝试：

Flux<Stuff> repeated = Mono
    .defer(manager::Subscribe)
    .repeat();