将嵌套的 for 循环转换为反应式 Flux答案

【问题标题】：Converting nested for loops to reactive Flux将嵌套的 for 循环转换为反应式 Flux
【发布时间】：2021-12-09 12:51:06
【问题描述】：

我有一个需要在一系列嵌套循环中的代码逻辑。这是示例：

String q = null;
for(C c in getC()){
    if(Valid(c)){
        String Id = c.Id;
        List<M> mList = getM(Id);
        for(M m in mList){
            q = Message.Format("query:{}",m);
            List<D> dList = getD(Id, q);
            sendDToKafka(dList);
        }
    }
}

我正在尝试使用项目反应器将上述逻辑转换为反应式。我到目前为止的代码：

Flux.fromIterable(getC())
    .map(c -> c.getId)
    .doOnNext(cId -> getM(cId))
    .map(m -> m.trim())
    .doOnNext(m -> getD()) // need to pass in query and Id to getD()
    .subscribe();

我面临的问题很少：

如何将 IsValid() 方法合并到查询中。
我需要在两个地方重用我在第一张地图 - .map(c -> c.getId) 上获得的 cId 值。如果不立即在下一步中使用，我如何跟踪该值。
有没有办法在响应式查询中形成 q 变量以作为参数传递给 getD()
如果代码是一种有效的方法，我将非常感谢任何反馈。

【问题讨论】：

标签： java spring-boot reactive-programming project-reactor

【解决方案1】：

首先，doOnNext 方法是为了产生副作用，而不是为了改变事件的流程。此外，如果您要将某些内容转换为反应式，则整个管道需要是非阻塞的，并且您应该避免调用任何阻塞的代码。如果您有无法更改的阻塞代码，you can follow the advice here:

对于过滤，您可以使用filter 和在多个地方使用cId，或者将其作为元组传递到链中（那里有很多库），或者您可以为此创建自己的类。

【讨论】：

【解决方案2】：

如何将 IsValid() 方法合并到查询中。

有一个操作符：

Flux.fromIterable(getC())
            .filter(c -> valid(c))

我需要在两个地方重用我在第一张地图 - .map(c -> c.getId) 上获得的 cId 值。如果没有，我如何跟踪该值立即用于下一步。

在特别简单的情况下，您可以像这样简单地使用嵌套的flatmaps：

.flatMap(id ->
        Flux.fromIterable(getM(id))
            .flatMap(m -> {
                String q = Message.Format("query: {}", m);
                List<D> dList = getD(id, q);
                return sendDToKafka(dList);
            })
    )

见What is a good idiom for nested flatMaps in Java Reactor?

有没有办法在响应式查询中形成 q 变量以作为参数传递给 getD()

您不能在流中使用非最终的q 变量。看看AtomicReference<String>。

正如另一个答案中已经提到的，doOnNext 用于副作用，map 用于将某些东西从一种类型映射到另一种类型。

【讨论】：