Spring WebFlux-将Flux转换为List

Question

我正在学习Spring WebFlux。

我的实体是这样的：

@Table("users")
public class User {
    @Id
    private Integer id;
    private String name;
    private int age;
    private double salary;
}

我有一个储存库(R2使用H2数据库)如下:

public interface UserRepository extends ReactiveCrudRepository<User,Integer> {
   
}

我的控制器是：

    @Autowired
    private UserRepository userRepository;

    private static List<User> userList = new ArrayList<>();

    @PostConstruct
    public void initializeStockObjects() {
        User stock1 = new User(11, "aaaa", 123, 123);
        User stock2 = new User(12, "bbb", 123, 123);
        User stock3 = new User(13, "ccc", 123, 123);
        userList.add(stock1);
        userList.add(stock2);
        userList.add(stock3);
    }

    @RequestMapping(value = "/livelistofusers", method = RequestMethod.GET, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<List<User>> getUsers() {
        return getUserData(userList);
    }

    public Flux<List<User>> getUserData(List<User> userList) {
       Flux<Long> interval = Flux.interval(Duration.ofSeconds(3));
       interval.subscribe((i) -> userList.forEach(user -> addNewUser(user)));
       Flux<List<User>> transactionFlux = Flux.fromStream(Stream.generate(() -> userList));
       return Flux.zip(interval, transactionFlux).map(Tuple2::getT2);
     }

目前为止一切正常。我能够每3秒钟向视图返回整个用户列表。这里没有任何问题。

现在，我想将Flue，即Flux Flux2=userRepository.findAll（）发送到视图。这意味着，不是返回getUserData（userList）；如何才能返回getUserData（Flux2（...我应该在这里做什么？？？...我尝试了几件事，但我最终制作了阻塞列表而不是非阻塞...）；？

问题：我如何才能做到这一点？即我如何才能每3秒将整个Flux发送到我的视图。我在这里感到迷茫和无知。任何相关的帮助链接或解决方案将不胜感激。

编辑:

根据尼普纳的评论，我尝试了一下：

@RequestMapping(value = "/livelistofusersall", method = RequestMethod.GET, produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<List<User>>  getUsersall() {
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(3));
        interval.subscribe((i) -> userRepository.findAll());
        Flux<List<User>> transactionFlux = userRepository.findAll().collectList().flatMapMany(Flux::just);
        return Flux.zip(interval, transactionFlux).map(Tuple2::getT2);
    }

但是现在在我的上下文路径中，列表在等待3秒钟后“只加载一次”。我错过了什么？

Answer 1

为此，您可以在Flux中使用< code>collectList()运算符，这将给出一个单一的列表。

userRepository.findAll().collectList().flatMapMany(Flux::just);