import org.junit.Test;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import reactor.core.publisher.Flux;
|
import reactor.core.publisher.Mono;
|
import reactor.core.scheduler.Schedulers;
|
|
import java.util.Iterator;
|
|
public class ReactorTest {
|
|
private static final Logger LOGGER = LoggerFactory.getLogger(ReactorTest.class);
|
|
|
/**
|
* 随便测试下
|
*/
|
@Test
|
public void concurrentTest(){
|
|
//这里没有什么用,纯粹是Schedulers.elastic()可以复用这里的线程池,不想写多的代码了
|
Flux.range(1,100)
|
.map(a -> a*1)
|
.subscribeOn(Schedulers.elastic())
|
.subscribe(System.out::println);
|
|
//开始测试了
|
long start = System.currentTimeMillis();
|
|
|
//第一个参数20 20个并发
|
//后面表示N个请求,最长的一个请求可能要2000ms
|
list(20, 1000l,2000l,100l,200l,300l,400l,500l,600l,700l,800l,900l)
|
.forEachRemaining( show -> LOGGER.info(show) );
|
|
LOGGER.info("总时间 : {} ms", System.currentTimeMillis() - start );
|
|
}
|
|
/**
|
* 并行执行
|
* @param concurrent 并行数量
|
* @param sleeps 模拟停顿时间
|
* @return 随便返回了
|
*/
|
public Iterator<String> list(int concurrent, Long... sleeps){
|
return Flux.fromArray(sleeps)
|
.log()
|
.flatMap(sleep -> Mono.fromCallable( () -> mockHttp(sleep)).subscribeOn(Schedulers.elastic()), concurrent)
|
.toIterable().iterator();
|
}
|
|
/**
|
* 实际上是一个http请求
|
* @param sleep 请求耗时
|
* @return
|
*/
|
public String mockHttp(long sleep){
|
try {
|
Thread.sleep(sleep);
|
LOGGER.info("停顿{}ms真的执行了", sleep);
|
} catch (InterruptedException e) {
|
e.printStackTrace();
|
}
|
return String.format("停顿了%sms", sleep);
|
}
|
|
}
|
