如何使用 Spring Webflux 创建列表流？答案

【问题标题】：How to create a stream of lists using Spring Webflux?如何使用 Spring Webflux 创建列表流？
【发布时间】：2018-11-14 00:30:26
【问题描述】：

我在 Spring Boot 项目中使用 Spring Webflux 和 ReactiveMongoRepository，我希望每个通量都是一个对象列表，例如：

// 1st Flux Started:
[
{
// 1st Dashboard
},
{
// 2nd Dashboard
}
]
// 1st Flux Ended
// 2nd Flux Started:
[
{
// 1st Dashboard
},
{
// 2nd Dashboard
}
]
// 2nd Flux Ended

存储库：

@Repository
public interface ReactiveDashboardRepository extends ReactiveMongoRepository<Dashboard, String> {
}

服务：

@Service
public class ReactiveDashboardServiceImpl implements ReactiveDashboardService {

    private ReactiveDashboardRepository reactiveDashboardRepository;

    public ReactiveDashboardServiceImpl(ReactiveDashboardRepository reactiveDashboardRepository) {
        this.reactiveDashboardRepository = reactiveDashboardRepository;
    }

    public Flux<Dashboard> getDashboards() {

        return this.reactiveDashboardRepository.findAll();
    }


}

控制器：

@CrossOrigin
@RestController
@RequestMapping("/api/sse")
public class ReactiveDashboardRestController {

    private ReactiveDashboardService reactiveDashboardService;

    public ReactiveDashboardRestController(ReactiveDashboardService reactiveDashboardService) {
        this.reactiveDashboardService = reactiveDashboardService;
    }

    @GetMapping(value = "/dashboards", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Dashboard> getDashboards() {
        return this.reactiveDashboardService.getDashboards();
    }

}

所以基本上我希望数组中的所有对象都成为单个流的一部分，因为这就是我将使用的客户端应用程序的设计方式。目前总共只有3个。所以在每个流中都会保存一个所有对象的数组。我知道这不是 webflux 的最佳用途。如果可以实现，如何实现？

【问题讨论】：

您要将哪些数据转换为该表单？更具体一些，并给出一些代码。
我通过添加额外信息来编辑我的问题。
它们在 Dashboard 类中是否有任何属性，可以使用哪些属性对通量进行分组？如果是，那么您可以将类似的通量组合在一起，最终获得仪表板列表的通量。

标签： spring-webflux

【解决方案1】：

如果你想要一个无限 Flux 以间隔发送数据，你需要函数“interval”。

为了简化您的代码，我建议您直接在 List 中返回 Dashboards，并在 Controller 中的 Flux 中转换 List。

会是这样的：

    List<Dashboard> lDashBoard = new ArrayList<Dashboard>(Arrays.asList(
            Dashboard.builder().name("Dash1").build(),
            Dashboard.builder().name("Dash2").build(),
            Dashboard.builder().name("Dash3").build()
            ));
    Flux<List<Dashboard>> intervalFlux = Flux
            .interval(Duration.ofMillis(500))
            .map(tick -> {
                return lDashBoard;
            });
    intervalFlux.subscribe(t->log.info("Dashboards:{}", lDashBoard));

输出如下：

2018-11-14 15:55:18.575  INFO [Dashboards:[Dashboard(name=Dash1), Dashboard(name=Dash2), Dashboard(name=Dash3)]
2018-11-14 15:55:19.073  INFO [Dashboards:[Dashboard(name=Dash1), Dashboard(name=Dash2), Dashboard(name=Dash3)]
2018-11-14 15:55:19.573  INFO [Dashboards:[Dashboard(name=Dash1), Dashboard(name=Dash2), Dashboard(name=Dash3)]

【讨论】：

我的想法是在一个流中将所有项目推送到客户端，每个流之间的间隔为 30 秒。通量流将是无限的流。你知道如何解决这个问题吗？一般来说，我对反应式编程真的很陌生，即使基于客户端逻辑的解决方案没有意义，我也必须在我的项目中应用类似的东西。