@Test
void parallelHttpRequests() {
// this is just for limiting the test running period otherwise you don't need it
int COUNT = 2;
// use whatever (blocking) http client you desire;
// when using e.g. WebClient (Spring, non blocking client)
// the example will slightly change for no longer use
// subscribeOn(Schedulers.elastic())
RestTemplate client = new RestTemplate();
// exit, lock, condition are provided to allow one to run
// all this code in a @Test, otherwise they won't be needed
var exit = new AtomicBoolean(false);
var lock = new ReentrantLock();
var condition = lock.newCondition();
MessageFormat message = new MessageFormat("#batch: {0}, #req: {1}, resultLength: {2}");
Flux.interval(Duration.ofSeconds(1L))
.take(COUNT) // this is just for limiting the test running period otherwise you don't need it
.doOnNext(batch -> debug("#batch", batch)) // just for debugging
.flatMap(batch -> Flux.range(1, 10) // 10 requests per 1 second
.flatMap(i -> Mono.fromSupplier(() ->
client.getForEntity("https://www.google.com/", String.class).getBody()) // your request goes here (1 of 10)
.map(s -> message.format(new Object[]{batch, i, s.length()})) // here the request's result will be the output of message.format(...)
.doOnSubscribe(s -> debug("doOnSubscribe: #batch = " + batch + ", i = " + i)) // just for debugging
.subscribeOn(Schedulers.elastic()) // one I/O thread per request
)
)
// consider using onErrorResume to handle overflow IllegalStateException
.subscribe(
s -> debug("received", s) // do something with the above request's result
e -> {
// pay special attention to overflow IllegalStateException
debug("error", e.getMessage());
signalAll(exit, condition, lock);
},
() -> {
debug("done");
signalAll(exit, condition, lock);
}
);
await(exit, condition, lock);
}
// you won't need the "await" and "signalAll" methods below which
// I created only to be easier for one to run this in a test class
private void await(AtomicBoolean exit, Condition condition, Lock lock) {
lock.lock();
while (!exit.get()) {
try {
condition.await();
} catch (InterruptedException e) {
// maybe spurious wakeup
e.printStackTrace();
}
}
lock.unlock();
debug("exit");
}
private void signalAll(AtomicBoolean exit, Condition condition, Lock lock) {
exit.set(true);
try {
lock.lock();
condition.signalAll();
} finally {
lock.unlock();
}
}
2条答案
按热度按时间4sup72z81#
您可以使用
delayElements
而不是整个zipwith
。9gm1akwq2#
我们可以使用
Flux.delayElements
每1秒处理一批10个请求;请注意,如果处理时间超过1秒,下一批仍将并行启动,因此将与前一批一起处理(可能还有许多其他前一批)!这就是为什么我提出了另一种解决方案,其中10个请求的批处理仍然每1秒处理一次,但是如果其处理时间超过1秒,则下一批将失败(参见
overflow
IllegalStateException
);可以处理该故障,以便继续整个处理,但我不会在这里显示,因为我希望保持示例简单;请参阅onErrorResume
对处理overflow
IllegalStateException
的有用性。下面的代码将以每秒10个请求的速率在
https://www.google.com/
上执行GET
。您必须做一些额外的修改,以支持服务器无法在1秒内处理所有10个请求的情况;当你的服务器还在处理前一秒发出的请求时,你可以跳过发送请求。