java - Spring 5 FluxSink执行fluxSink.next时不发送数据

Question

我有一个生成 Randon INT 并发送回客户端的 GET 方法示例

@GetMapping
public Flux<String> search() {
    return Flux.create(fluxSink -> {

            Random r = new Random();
            int n;
            for (int i = 0; i < 10; i++) {
                n= r.nextInt(1000);
                System.out.println("Creating:"+n);
                fluxSink.next(String.valueOf(n));

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            fluxSink.complete();


    });
}

curl :

curl -v -H "Accept: text/event-stream" -X GET 'http://localhost:8080/stream

使用此代码使用 CURL 命令时，我只能在 fluxSink.complete 发生时在我的客户端中看到数字。

现在如果我将我的代码更改为:

@GetMapping
public Flux<String> search() {
    return Flux.create(fluxSink -> {
        Thread t = new Thread(() -> {
            Random r = new Random();
            int n;
            for (int i = 0; i < 10; i++) {
                n= r.nextInt(1000);
                System.out.println("Creating:"+n);
                fluxSink.next(String.valueOf(n));

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            fluxSink.complete();
        });
        t.start();
    });
}

将流程包装到线程中，效果很好。当 fluxSink.next 发生时，我可以看到数据传输正常。

谁能解释一下这个效果？如何在不显式使用线程的情况下查看数据流动？

谢谢!

Answer 1

Flux.create() 是关于“通过 FluxSink API 以同步或异步方式发射多个元素”。在您的情况下，您正在同步发射所有元素，并且您没有机会让 Reactor 工作，因为在完成所有操作之前您永远不会退出该方法。简而言之，您正在阻塞当前线程，这是 Reactor 所禁止的。

使用线程让 Reactor 有机会分派(dispatch)这些信号。但 Flux.create() 的精神(如其 javadoc 所示)是关于适应回调。

在这种情况下，Flux.generate 可能更合适。尝试用 Thread 模拟延迟不是一个好主意。您可以改为执行以下操作:

Flux<String> response = Flux.just(aListOfElements).delayElements(Duration.ofSeconds(2));