gpt4 book ai didi

java - 通过 SSE 订阅的 Flux 引发 cancel() 事件

转载 作者:行者123 更新时间:2023-12-05 06:36:54 28 4
gpt4 key购买 nike

我有一个 Spring Boot 2.0.0.M7 + Spring Webflux 应用程序,我在其中使用 Thymeleaf Reactive。

我注意到在我的微服务上,当我调用端点以 SSE 模式(文本/事件流)返回数据流时,即使已正确处理该数据流也会发生 cancel()。

例如,这是一个简单的 Controller 端点:

@GetMapping(value = "/posts")
public Flux<String> getCommunityPosts() {
return Flux.just("A", "B", "C").log("POSTS");
}

这是我在 SSE 模式下请求时获得的订阅流量日志:

2018-02-13 17:04:09.841  INFO 4281 --- [nio-9090-exec-4] POSTS : | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
2018-02-13 17:04:09.841 INFO 4281 --- [nio-9090-exec-4] POSTS : | request(1)
2018-02-13 17:04:09.842 INFO 4281 --- [nio-9090-exec-4] POSTS : | onNext(A)
2018-02-13 17:04:09.847 INFO 4281 --- [nio-9090-exec-4] POSTS : | request(1)
2018-02-13 17:04:09.847 INFO 4281 --- [nio-9090-exec-4] POSTS : | onNext(B)
2018-02-13 17:04:09.848 INFO 4281 --- [nio-9090-exec-4] POSTS : | request(1)
2018-02-13 17:04:09.848 INFO 4281 --- [nio-9090-exec-4] POSTS : | onNext(C)
2018-02-13 17:04:09.849 INFO 4281 --- [nio-9090-exec-4] POSTS : | request(1)
2018-02-13 17:04:09.849 INFO 4281 --- [nio-9090-exec-4] POSTS : | onComplete()
2018-02-13 17:04:09.852 INFO 4281 --- [nio-9090-exec-4] POSTS : | cancel()

我们可以注意到 onComplete 之后的 cancel 事件。当我通过经典 GET 请求调用同一个端点时,我没有这种行为。我怀疑这个取消事件会使客户端事件源 (javascript) 抛出一个 onError 事件。

这是特定于 SSE 的已知/想要的行为吗?

问题更新

我实际上在我的一些流中使用 SSE,因为有时我需要我的事件源来获取 JSON 数据,而不是 Thymeleaf 已经处理过的 HTML。我应该用另一种方式做吗?

我的实现基于此示例的最后一个方法:https://github.com/danielfernandez/reactive-matchday/blob/master/src/main/java/com/github/danielfernandez/matchday/web/controller/MatchController.java

但是,我可能错过了在我之前的帖子中提供的一些信息。我使用 Tomcat 服务器(8.5.23 和 M7),而不是 Netty 服务器。我强制 Tomcat 使用包括以下 Maven 依赖项:

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
</dependency>

在示例项目中使用您的代码,这似乎会导致问题。

当我在 Netty 服务器上运行代码时,我得到的结果与您相同:

2018-02-14 12:30:48.713  INFO 3060 --- [ctor-http-nio-2] reactor.Flux.ConcatMap.1 : onSubscribe(FluxConcatMap.ConcatMapImmediate)
2018-02-14 12:30:48.714 INFO 3060 --- [ctor-http-nio-2] reactor.Flux.ConcatMap.1 : request(1)
2018-02-14 12:30:49.717 INFO 3060 --- [ parallel-2] reactor.Flux.ConcatMap.1 : onNext(a)
2018-02-14 12:30:49.739 INFO 3060 --- [ctor-http-nio-2] reactor.Flux.ConcatMap.1 : request(31)
2018-02-14 12:30:50.731 INFO 3060 --- [ parallel-3] reactor.Flux.ConcatMap.1 : onNext(b)
2018-02-14 12:30:51.733 INFO 3060 --- [ parallel-4] reactor.Flux.ConcatMap.1 : onNext(c)
2018-02-14 12:30:51.735 INFO 3060 --- [ parallel-4] reactor.Flux.ConcatMap.1 : onComplete()

当我在 Tomcat 服务器上运行相同的代码时,我遇到了取消问题:

2018-02-14 12:33:18.294  INFO 3088 --- [nio-8080-exec-3] reactor.Flux.ConcatMap.2 : onSubscribe(FluxConcatMap.ConcatMapImmediate)
2018-02-14 12:33:18.295 INFO 3088 --- [nio-8080-exec-3] reactor.Flux.ConcatMap.2 : request(1)
2018-02-14 12:33:19.295 INFO 3088 --- [ parallel-4] reactor.Flux.ConcatMap.2 : onNext(a)
2018-02-14 12:33:19.297 INFO 3088 --- [ parallel-4] reactor.Flux.ConcatMap.2 : request(1)
2018-02-14 12:33:20.302 INFO 3088 --- [ parallel-5] reactor.Flux.ConcatMap.2 : onNext(b)
2018-02-14 12:33:20.302 INFO 3088 --- [ parallel-5] reactor.Flux.ConcatMap.2 : request(1)
2018-02-14 12:33:21.306 INFO 3088 --- [ parallel-6] reactor.Flux.ConcatMap.2 : onNext(c)
2018-02-14 12:33:21.306 INFO 3088 --- [ parallel-6] reactor.Flux.ConcatMap.2 : request(1)
2018-02-14 12:33:21.307 INFO 3088 --- [ parallel-6] reactor.Flux.ConcatMap.2 : onComplete()
2018-02-14 12:33:21.307 INFO 3088 --- [nio-8080-exec-4] reactor.Flux.ConcatMap.2 : cancel()

这可能是 Tomcat 问题还是我做错了什么?

最佳答案

首先,我认为您不应该将 SSE 用于有限流。

当我创建一个 Controller 方法时:

@GetMapping(path = "/test", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
@ResponseBody
public Flux<String> test() {
return Flux.just("a", "b", "c").delayElements(Duration.ofSeconds(1)).log();
}

并通过浏览器(Chrome 或 Firefox)请求它:

<script type="text/javascript">
var testEventSource = new EventSource("/test");
testEventSource.onmessage = function (e) {
console.log(e);
};
</script>

我在服务器上得到以下日志:

| onSubscribe([Fuseable] FluxOnAssembly.OnAssemblySubscriber)
| request(1)
| onNext(a)
| request(31)
| onNext(b)
| onNext(c)
| onComplete()
| onSubscribe([Fuseable] FluxOnAssembly.OnAssemblySubscriber)
| request(1)
| onNext(a)
| request(31)
| onNext(b)
| onNext(c)
| onComplete()

一旦 Flux 完成,服务器就会关闭连接,浏览器会自动重新连接。这将一遍又一遍地重播相同的序列。

我在服务器上获得 cancel() 事件的唯一方法是在流期间关闭浏览器选项卡。

关于java - 通过 SSE 订阅的 Flux 引发 cancel() 事件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48771125/

28 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com