使用 Redis 的 SSE 端点的奇怪行为答案

【问题标题】：Strange behavior with SSE endpoint using Redis使用 Redis 的 SSE 端点的奇怪行为
【发布时间】：2019-07-17 08:00:19
【问题描述】：

如果在 Redis 中，我需要将一些数据推送到客户端，但客户端每 5 秒不断重新连接到 SSE 端点。

后端代码：

@RestController
@RequestMapping("/reactive-task")
public class TaskRedisController {

    private final TaskRedisRepository taskRedisRepository;

    TaskRedisController(TaskRedisRepository taskRedisRepository){
        this.taskRedisRepository = taskRedisRepository;
    }

   @CrossOrigin
    @GetMapping(value = "/get/{id}")
    public Flux<ServerSentEvent<Task>> getSseStream2(@PathVariable("id") String id) {
        return taskRedisRepository.findByTaskId(id)
            .map(task -> ServerSentEvent.<Task>builder().data(task).build());
    }
}

@Repository
public class TaskRedisRepository {
    public Flux<Task> findByTaskId(String id) {
        return template.keys("task:" + id).flatMap(template.opsForValue()::get);
    } 
}


@AllArgsConstructor
@NoArgsConstructor
@Getter
@Setter
@EqualsAndHashCode
@Entity
public class Task {
    @Id
    @GeneratedValue(strategy=GenerationType.IDENTITY)
    private Long id;
    @Column(length = 25)
    private String result;
}

客户端使用JS消费：

var evt = new EventSource("http://localhost:8080/reactive-task/get/98"); evt.onmessage = function(event) {
  console.log(event);
};

谁能指出我正确的方向？
任何建议将不胜感激。

更新：我需要在 Redis 中存储数据一段时间（5-10 分钟）。更新：我在 MongoDB 上写过类似的代码，效果很好。

【问题讨论】：

标签： spring redis spring-webflux server-sent-events spring-data-redis-reactive

【解决方案1】：

在这种情况下，taskRedisRepository.findByTaskId(id) 可能发送一个有限的Flux - 表示几个元素和一个完整的信号完成流。

Spring WebFlux 会将onComplete 信号解释为流的结尾并关闭它。浏览器 SSE 客户端的默认行为是在连接终止时立即重新连接到端点。

如果您希望保持持久连接并且只在添加新元素时收到通知，则需要直接将其作为数据存储的一项功能加以利用。对于 Redis，发布/订阅功能 (see reference documentation) 支持此模式。

总而言之，我认为您看到了预期的行为，因为在这种情况下，您的数据存储不会生成无限流，通知您添加到集合中的新元素，而是该集合中存在的有限元素流给定的时间。

【讨论】：