我想处理Flux
,以限制由Mono
列表发出的并发HTTP请求数量。
当一些请求完成(收到响应)后,服务再发送一个请求,直到等待请求的总数为15。
单个请求返回一个列表,并根据结果触发另一个请求。
现在,我想限制并发地发送请求。因为消费方,太多的HTTP请求会让服务器陷入麻烦。
我使用以下方式的flatMapMany
。
public Flux<JsonNode> syncData() {
return service1
.getData(param1)
.flatMapMany(res -> {
List<Mono<JsonNode>> totalTask = new ArrayList<>();
Map<String, Object> originData = service2.getDataFromDB(param2);
res.withArray("data").forEach(row -> {
String id = row.get("id").asText();
if (originData.containsKey(id)) {
totalTask.add(service1.updateRequest(param3));
} else {
totalTask.add(service1.deleteRequest(param4));
}
originData.remove(id);
});
for (left) {
totalTask.add(service1.createRequest(param5));
}
return Flux.merge(totalTask);
});
}
void syncData() {
syncDataService.syncData().????;
}
我试过链式使用 .window(15)
,但它并不起作用。所有请求都同时发送。
为了实现我的目标,我该如何处理 Flux
?