使用 SpringFlux 的 webclient 重复 Mono
Repeat Mono with webclient from SpringFlux
情况如下:
我发送了第一个请求,然后以 5s 的间隔重复发送了第二个请求。如果第二个请求获得成功响应,我将其保存在数据库中并执行一些操作,如果它获得不成功(完成)响应,则应重复请求。当尝试次数超过 3 次时,我需要停止重复第二个请求。我该怎么做?
return firstRequestSenderService.send(request)
.flatMap(resp -> {
AtomicInteger attempts = new AtomicInteger(0);
String url = normalizeUrl(resp.getResult());
return Mono.defer(() -> {
log.info("Second request, attempt = {}, params = {}", attempts.get(), param);
return secondRequestSenderService.send(param, url, attempts.getAndIncrement());
})
.filter(this::isCompleteResponse)
// i try .filter(b -> attempts.get() > 2)
.doOnNext(r -> log.info("Save report"))
.map(secondResp -> dataSaver.saveReport(param, secondResp))
.doOnNext(r -> log.info("Send request to another service"))
.flatMap(r -> secondRequestSender.sendPdf(r)))
.doOnNext(bytes -> dataSaver.saveAnotherReport(param, bytes))
.repeatWhenEmpty(req -> Flux.interval(Duration.ofSeconds(5)));
// also try .repeatWhenEmpty(3, req -> Flux.interval(Duration.ofSeconds(5)));
})
.then(Mono.empty());
检查reactor-addons
你可以做类似的事情
return firstRequestSenderService.send(request)
.flatMap(resp -> {
return secondRequestSenderService
.send(param, url, attempts.getAndIncrement())
.retryWhen(Retry.any()
.randomBackoff(Duration.ofMillis(100), Duration.ofMillis(5000))
.retryMax(3));
});
你也可以用 reactor core retry utils 做到这一点 RetryBackoffSpec 当你使用最新的 spring webflux
时你会得到它
RetryBackoffSpec retryBackoffSpec = Retry
.fixedDelay(3, Duration.ofSeconds(5));
return firstRequestSenderService.send(request)
.flatMap(resp -> {
return secondRequestSenderService
.send(param, url, attempts.getAndIncrement())
.retryWhen(retryBackoffSpec);
});
情况如下: 我发送了第一个请求,然后以 5s 的间隔重复发送了第二个请求。如果第二个请求获得成功响应,我将其保存在数据库中并执行一些操作,如果它获得不成功(完成)响应,则应重复请求。当尝试次数超过 3 次时,我需要停止重复第二个请求。我该怎么做?
return firstRequestSenderService.send(request)
.flatMap(resp -> {
AtomicInteger attempts = new AtomicInteger(0);
String url = normalizeUrl(resp.getResult());
return Mono.defer(() -> {
log.info("Second request, attempt = {}, params = {}", attempts.get(), param);
return secondRequestSenderService.send(param, url, attempts.getAndIncrement());
})
.filter(this::isCompleteResponse)
// i try .filter(b -> attempts.get() > 2)
.doOnNext(r -> log.info("Save report"))
.map(secondResp -> dataSaver.saveReport(param, secondResp))
.doOnNext(r -> log.info("Send request to another service"))
.flatMap(r -> secondRequestSender.sendPdf(r)))
.doOnNext(bytes -> dataSaver.saveAnotherReport(param, bytes))
.repeatWhenEmpty(req -> Flux.interval(Duration.ofSeconds(5)));
// also try .repeatWhenEmpty(3, req -> Flux.interval(Duration.ofSeconds(5)));
})
.then(Mono.empty());
检查reactor-addons 你可以做类似的事情
return firstRequestSenderService.send(request)
.flatMap(resp -> {
return secondRequestSenderService
.send(param, url, attempts.getAndIncrement())
.retryWhen(Retry.any()
.randomBackoff(Duration.ofMillis(100), Duration.ofMillis(5000))
.retryMax(3));
});
你也可以用 reactor core retry utils 做到这一点 RetryBackoffSpec 当你使用最新的 spring webflux
时你会得到它 RetryBackoffSpec retryBackoffSpec = Retry
.fixedDelay(3, Duration.ofSeconds(5));
return firstRequestSenderService.send(request)
.flatMap(resp -> {
return secondRequestSenderService
.send(param, url, attempts.getAndIncrement())
.retryWhen(retryBackoffSpec);
});