使用SpringBoot 2和Poi类（兴趣点）：
public class Poi {
public Poi(String poidId, Double price, Double latitude, Double longitude) {...}
private String poidId;
private Double latitude;
private Double longitude;
private Double price;
//And Getters and Setters
}
我有2个Poi通量：
Flux<Poi> availablePoisFlux;
Flux<Poi> poiFlux;
可用的第一个元素PoisFlux包含具有以下内容的Pois：
一个poidId
没有纬度信息
没有经度信息
价格信息
第二个元素poiFlux包含具有以下内容的Pois：
一个poidId
纬度
经度
没有价格信息
（poidId是Poi的标识符）。
我想从两个通量（poiFlux和availablePoisFlux）中使用Pois（具有poidId，价格，经度和纬度）创建一个新的Flux resultPoisFlux。
poidId属性是两个Flux（poiFlux和availablePoisFlux）之间的键。
样本实施：
我想我可以使用zipWith运算符来做到这一点，但是我需要有关反应式运算符（和filter？）的一些信息和建议。
我想使用poidId标识符从第一个通量进行迭代，并从第二个通量获取信息（价格），并使用正确的值更新price属性。
输入值样本：
poiFlux = Poi(poidId=poiId0, price=null, name=name0, latitude=2.2222, longitude=14.222)
poiFlux = Poi(poidId=poiId1, price=null, name=name1, latitude=3.2222, longitude=15.222)
poiFlux = Poi(poidId=poiId2, price=null, name=name2, latitude=4.2222, longitude=16.222)
poiFlux = Poi(poidId=poiId3, price=null, name=name3, latitude=5.2222, longitude=17.222)
poiFlux = Poi(poidId=poiId4, price=null, name=name4, latitude=6.2222, longitude=18.222)
poiFlux = Poi(poidId=poiId5, price=null, name=name5, latitude=7.2222, longitude=19.222)
poiFlux = Poi(poidId=poiId6, price=null, name=name6, latitude=8.2222, longitude=20.222)
poiFlux = Poi(poidId=poiId7, price=null, name=name7, latitude=9.2222, longitude=21.222)
poiFlux = Poi(poidId=poiId8, price=null, name=name8, latitude=10.2222, longitude=22.222)
poiFlux = Poi(poidId=poiId9, price=null, name=name9, latitude=11.2222, longitude=23.222)
availablePoisFlux = Poi(poidId=poiId0, price=120.0, name=name0, latitude=null, longitude=null)
availablePoisFlux = Poi(poidId=poiId1, price=120.0, name=name1, latitude=null, longitude=null)
availablePoisFlux = Poi(poidId=poiId2, price=120.0, name=name2, latitude=null, longitude=null)
availablePoisFlux = Poi(poidId=poiId3, price=120.0, name=name3, latitude=null, longitude=null)
availablePoisFlux = Poi(poidId=poiId4, price=120.0, name=name4, latitude=null, longitude=null)
预期结果：
resultPoisFlux = Poi(poidId=poiId0, price=120.0, name=name0, latitude=2.2222, longitude=14.222)
resultPoisFlux = Poi(poidId=poiId1, price=120.0, name=name1, latitude=3.2222, longitude=15.222)
resultPoisFlux = Poi(poidId=poiId2, price=120.0, name=name2, latitude=4.2222, longitude=16.222)
resultPoisFlux = Poi(poidId=poiId3, price=120.0, name=name3, latitude=5.2222, longitude=17.222)
resultPoisFlux = Poi(poidId=poiId4, price=120.0, name=name4, latitude=6.2222, longitude=18.222)
像这样：
Flux<Poi> resultPoisFlux = availablePoisFlux.zipWith(poiFlux, (a, b) -> new Poi(a.getPoidId(), a.getPrice(), getLatitudeFromPoiFluxByPoidId(a.getPoidId()), getLongitudeFromPoiFluxByPoidId(a.getPoidId())))....
感谢您的帮助。
最佳答案
zip/zipWith，但它只能成对组合两个来源...
...只要有足够的元素可以配对。因此，只有在保证两个源中的元素都以相同顺序排列并且两边的poiIds中没有差异时，这才有用。在您的示例中，是这样的，因为即使第二个源仅包含4个元素，这些元素也与第一个源的开头相同。
poiFlux.zipWith(availablePoisFlux, (a, b) -> new Poi(a.getPoiId(),
b.getPrice(),
a.getLatitude(),
a.getLongitude(),
a.getName()));
更通用的解决方案，反应更少
如果没有这样的保证，则需要以某种方式组合两个无序和不相交的序列。如果不收集一个源中的所有元素（最好是
availablePoisFlux），则无法做到这一点，这意味着它将延迟另一个源的处理，直到所述源完成为止。合并的一种方法是将所有值收集到由
poiId键控的映射中，然后在第二个源上“迭代”。由于可能在地图中找不到某些元素，因此需要handle才能“跳过”这些元素：availablePoisFlux.collectMap(Poi::getId, Poi::getPrice)
.flatMapMany(knownPrices -> poiFlux.handle((poi, sink) -> {
String poiId = poi.getPoiId();
if (knownPrices.containsKey(poiId) {
Double price = knownPrices.get(poiId);
Poi complete = new Poi(poiId, price, poi.getLatitude(),
poi.getLongitude(), poi.getName());
sink.next(complete);
} //else do nothing and let handle skip that poi
}));