小编Mar*_*nyi的帖子

我向第三方 Web 服务重复发出分页 WebClient 请求。我现在的实现可以工作，但正在阻塞。

到目前为止我的实现：

var elementsPerPage = 10;
Flux
    .generate(
        () -> 0,
        (pageIndex, emitter) -> {
            BlahServiceResponse blahServiceResponse =
                webClient
                    .get()
                    .uri("/blah?pageIndex={pageIndex}", pageIndex)
                    .retrieve()
                    .bodyToMono(BlahServiceResponse.class)
                    .block(); // Yuck!!!
            if (blahServiceResponse.getStudents().size() > 0) {
                emitter.next(blahServiceResponse);
            } else {
                emitter.complete();
            }
            return pageIndex + elementsPerPage;
        }
    )
    .subscribe(System.out::println); // Replace me with actual logic

出于可以理解的原因，如果上面的代码更改为以下内容，则会引发“IllegalStateException：生成器没有调用任何 SynchronousSink 方法”异常：

webClient
    .get()
    ...
    .bodyToMono(BlahServiceResponse.class)
    .subscribe(emitter::next);

所以我开始寻找一个异步接收器并意识到它是 Flux|MonoSink。但据我所知，Flux 中没有构建器方法支持使用 Flux|MonoSink 生成有状态元素。

我是否遗漏了一些东西，是否有更优雅的方法？

我在 Vert.x 应用程序中使用了一个库，它返回Project Reactor类型Mono。

我有一个 Verticle 接收这种反应类型，并打算通过事件总线将内容发送到另一个 Verticle：

import io.vertx.core.AbstractVerticle;
import io.vertx.core.eventbus.Message;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;

public class HelperVerticle extends AbstractVerticle
{
    public static final String ADDRESS = "address_1";

    @Override
    public void start() throws Exception
    {
        vertx.eventBus().consumer(ADDRESS, this::consume);
    }

    private void consume(Message<Object> message)
    {
        Mono.delay(Duration.ofMillis(3000)) 
            .thenReturn("Content of Mono.") // this would come from external library
            .publishOn(Schedulers.fromExecutor(vertx.nettyEventLoopGroup())) // is this needed?
            .subscribe(output ->
            {
                System.out.println("My verticle: " + Thread.currentThread().getName());
                message.reply(output + " " + message.body()); …

我有一个关于 Spring WebFlux 和 Reactor 的问题。我正在尝试编写一个简单的场景，其中在 GET 端点中，我返回代表实体的 DTO Flux，并且每个实体都有代表另一个实体的其他 DTO 的集合。这里请遵循详细信息。

我有两个实体，Person 和 Song，定义如下：

@Data
public class Person {
    @Id
    private Long id;
    private String firstName;
    private String lastName;
}

@Data
public class Song {
    @Id
    private Long id;
    private String title;
    private Long authorId;
}

这些实体由以下 DTO 表示：

@Data
public class SongDTO {
    private Long id;
    private String title;
    public static SongDTO from(Song s) {
        // converts Song to its dto
    }
}

@Data
public class PersonDTO { …