2

如果它在 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 上写了类似的代码，它工作正常。

4

1 回答 1

3

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

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

如果您希望保持持久连接并且只在添加新元素时收到通知，则需要直接利用它作为数据存储的功能。对于 Redis，发布/订阅功能支持此模式（请参阅参考文档）。

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

于 2019-07-17T08:53:21.173 回答