来自控制器的WebFlux异步响应

Question

我的任务是简单地制作一个控制器，当它们准备好时立即给我结果（下面的简单示例）

我想得到字符串的确切数量（例如1000个字符串，以某种方式为1秒）（实际上我需要得到func的结果，但为了简化任务，只是字符串）

因此，当我在控制器中收到一些请求时，我希望它能以这种方式在它们准备好后尽快给出答案（无需缓冲结果）：

1秒钟

“一些绳子”-

1秒钟

“另一个”-

1秒钟

“第三个”-

1000秒

“一些绳子”

......

"千弦"

这是我的代码：

@GetMapping(value = "/3", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> get3() {
        System.out.println("get3 start");
        Flux<String> result = Flux.fromStream(IntStream.range(1, 10).mapToObj(i -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "flux data--" + i;
        }));
        System.out.println("get3 end");
        return result;

    }

实际上在我的控制台里

立即“get3 start”和“get3 end”，但响应仅在所有字符串准备就绪后进行

我对这个任务的实际服务是类似的（但我在这里合并了2个通量），我得到的通量是由间隔形成的，所以我希望它在结果出现时立即给我结果

public Flux<AnswerCalculationDto> calculate(CalculationDto calculationDto){
        String checkMsg = checkCalculationDto(calculationDto);
        if(checkMsg.equals("Success")){//valid
            Long quantity = Long.parseLong(calculationDto.getQuantity());

            Flux<AnswerCalculationDto> firstFunc =  Flux.interval(interval)//func 1
                    .onBackpressureDrop()
                    .takeWhile((i)-> i < quantity)
                    .map((i)->new AnswerCalculationDto(i,1,translateToJava(calculationDto.getFunc1(),i)))
                    ;
            Flux<AnswerCalculationDto> secondFunc = Flux.interval(interval) //func 2
                    .onBackpressureDrop()
                    .takeUntil((i)-> i > quantity-2)
                    .map((i)->new AnswerCalculationDto(i,2,translateToJava(calculationDto.getFunc2(),i)) )
                    ;
            return Flux.merge(firstFunc,secondFunc);
        }
        else {//invalid data from client
            return Flux.just(new AnswerCalculationDto("",checkMsg));
        }

    }

Answer 1

我在寻找的是Http流，此外请注意，Safari以及邮递员和axios（js lib-我在我的前端部分使用它）不支持超文本传输协议流，所以你不能看到你的输出出现，只要一个结果准备就绪（只有1响应中的所有结果），尝试Chrome。

此外，如果您像我一样与前端部分作斗争，请尝试搜索SSE-server-send-event，例如：https://turkogluc.com/server-sent-events-with-spring-boot-and-reactjs/

希望这能有所帮助

Answer 2

使用WebFlux从服务器流式传输数据有几个选项：

• 服务器发送的事件推送单个事件（媒体类型：text/event-stream）
• 用换行符分隔的流事件（媒体类型：Application/x-ndjson）

下面是一个完整的示例，它公开了文本/事件流

@RestController
public class StreamingController {

    @GetMapping(produces = TEXT_EVENT_STREAM_VALUE)
    Flux<DataEntry> sse() {
        return stream();
    }

    @GetMapping(produces = APPLICATION_NDJSON_VALUE)
    Flux<DataEntry> ndjson() {
        return stream();
    }

    private Flux<DataEntry> stream() {
        return Flux.range(1, 1000)
                .delayElements(Duration.ofSeconds(1))
                .map(i -> new DataEntry(i, Instant.now()));
    }

    @Value
    @Builder
    private static class DataEntry {
        long index;
        Instant timestamp;
    }
}

要测试文本/事件流使用：

curl-v-H“接受：文本/事件流”http://localhost:8080

> GET / HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: text/event-stream
> 
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< Content-Type: text/event-stream;charset=UTF-8
< 
data:{"index":1,"timestamp":"2022-04-08T14:41:06.513352Z"}

data:{"index":2,"timestamp":"2022-04-08T14:41:07.527817Z"}

data:{"index":3,"timestamp":"2022-04-08T14:41:08.541706Z"}

data:{"index":4,"timestamp":"2022-04-08T14:41:09.553329Z"}

要测试应用程序/x-ndjson，请使用：

curl-v-H“接受：应用程序/x-ndjson”http://localhost:8080

> GET / HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: application/x-ndjson
>
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< Content-Type: application/x-ndjson
< 
{"index":1,"timestamp":"2022-04-08T14:42:36.081269Z"}
{"index":2,"timestamp":"2022-04-08T14:42:37.094928Z"}
{"index":3,"timestamp":"2022-04-08T14:42:38.109378Z"}
{"index":4,"timestamp":"2022-04-08T14:42:39.121315Z"}

上述示例将以1秒的间隔生成1000条记录。您还可以使用以下方法生成无限流：

private Flux<DataEntry> stream() {
    return Flux.interval(Duration.ofSeconds(1))
            .map(i -> new DataEntry(i, Instant.now()));
}