响应式编程和协程在Java语言的应用

userService.getFavorites(userId, new Callback<List<String>>() {   public void onSuccess(List<String> list) {     if (list.isEmpty()) {       suggestionService.getSuggestions(new Callback<List<Favorite>>() {        public void onSuccess(List<Favorite> list) {           UiUtils.submitOnUiThread(() -> {             list.stream()                .limit(5)                .forEach(uiList::show);             });        }                public void onError(Throwable error) {           UiUtils.errorPopup(error);        }      });    } else {      list.stream()           .limit(5)          .forEach(favId -> favoriteService.getDetails(favId,             new Callback<Favorite>() {              public void onSuccess(Favorite details) {                UiUtils.submitOnUiThread(() -> uiList.show(details));              }
              public void onError(Throwable error) {                UiUtils.errorPopup(error);              }            }          ));    }  }
  public void onError(Throwable error) {    UiUtils.errorPopup(error);  }}

userService.getFavorites(userId)            .flatMap(favoriteService::getDetails)            .switchIfEmpty(suggestionService.getSuggestions())            .take(5)            .publishOn(UiUtils.uiThreadScheduler())            .subscribe(uiList::show, UiUtils::errorPopup);

userService.getFavorites(userId)           .timeout(Duration.ofMillis(800))            .onErrorResume(cacheService.cachedFavoritesFor(userId))            .flatMap(favoriteService::getDetails)            .switchIfEmpty(suggestionService.getSuggestions())           .take(5)           .publishOn(UiUtils.uiThreadScheduler())           .subscribe(uiList::show, UiUtils::errorPopup);

CompletableFuture<List<String>> ids = ifhIds(); 
CompletableFuture<List<String>> result = ids.thenComposeAsync(l -> {   Stream<CompletableFuture<String>> zip =      l.stream().map(i -> {         CompletableFuture<String> nameTask = ifhName(i);         CompletableFuture<Integer> statTask = ifhStat(i); 
        return nameTask.thenCombineAsync(statTask, (name, stat) -> "Name " + name + " has stats " + stat);       });  List<CompletableFuture<String>> combinationList = zip.collect(Collectors.toList());   CompletableFuture<String>[] combinationArray = combinationList.toArray(new CompletableFuture[combinationList.size()]);
  CompletableFuture<Void> allDone = CompletableFuture.allOf(combinationArray);   return allDone.thenApply(v -> combinationList.stream()      .map(CompletableFuture::join)       .collect(Collectors.toList()));});
List<String> results = result.join(); assertThat(results).contains(    "Name NameJoe has stats 103",    "Name NameBart has stats 104",    "Name NameHenry has stats 105",    "Name NameNicole has stats 106",    "Name NameABSLAJNFOAJNFOANFANSF has stats 121");

Flux<String> ids = ifhrIds(); 
Flux<String> combinations =    ids.flatMap(id -> {       Mono<String> nameTask = ifhrName(id);       Mono<Integer> statTask = ifhrStat(id); 
      return nameTask.zipWith(statTask,           (name, stat) -> "Name " + name + " has stats " + stat);    });
Mono<List<String>> result = combinations.collectList(); 
List<String> results = result.block(); assertThat(results).containsExactly(     "Name NameJoe has stats 103",    "Name NameBart has stats 104",    "Name NameHenry has stats 105",    "Name NameNicole has stats 106",    "Name NameABSLAJNFOAJNFOANFANSF has stats 121");

Flux<String> seq1 = Flux.just("foo", "bar", "foobar");Flux<String> seq2 = Flux.fromIterable(Arrays.asList("foo", "bar", "foobar"));Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3); 
Mono<String> noData = Mono.empty(); Mono<String> data = Mono.just("foo");

<dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-webflux</artifactId></dependency>

@RestController@RequestMapping(value = "/dsp")public class Controller {
    @Autowired    private StrategyService strategyService;
    @RequestMapping(value = "/routeA", method = RequestMethod.POST)    public ExampleResponse routeA(@RequestBody Map<String, String> requestBody) {      return strategyService.methodA();    }      @RequestMapping(value = "/routeB", method = RequestMethod.POST)    public ExampleResponse routeB(@RequestBody Map<String, String> requestBody) {      return strategyService.methodB();    }      @RequestMapping(value = "/routeC", method = RequestMethod.POST)    public ExampleResponse routeC(@RequestBody Map<String, String> requestBody) {      return strategyService.methodC();    }
}

@Configurationpublic class Router {
    @Bean    public RouterFunction<ServerResponse> route(StrategyHandler strategyHandler) {        return RouterFunctions            .route(RequestPredicates.POST("/dsp/routeA")                .and(RequestPredicates.accept(MediaType.APPLICATION_STREAM_JSON)), strategyHandler::methodA)            .andRoute(RequestPredicates.POST("/dsp/routeB")                .and(RequestPredicates.accept(MediaType.APPLICATION_STREAM_JSON)), strategyHandler::methodB)            .andRoute(RequestPredicates.POST("/dsp/routeC")                .and(RequestPredicates.accept(MediaType.APPLICATION_STREAM_JSON)), strategyHandler::methodC);    }    }

public Mono<ServerResponse> methodA(ServerRequest serverRequest) {        return ServerResponse.ok()                         .contentType(MediaType.APPLICATION_STREAM_JSON)                         .body(exampleService.dealWithRequest(                                 serverRequest.bodyToMono(String.class), serverRequest),                                                                                   String.class);}

Mono<String> response = null;
switch (method) {
    case GET:        response = WebClient.create()                            .method(HttpMethod.GET)                            .uri(requestUrl)                            .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)                            .retrieve()                            .bodyToMono(String.class);
        break;
    case POST:        response = WebClient.create()                            .method(HttpMethod.POST)                            .uri(requestUrl)                            .body(Mono.just(requestBody), String.class)                            .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)                            .headers(headers -> headers.addAll(httpHeaders))                            .retrieve()                            .bodyToMono(String.class);
        break;
    default:        break;}

@Beanpublic WebClient webClient(){    HttpClient httpClient = HttpClient.create()            .tcpConfiguration(client ->                    client.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 100)                    .doOnConnected(conn -> conn                            .addHandlerLast(new ReadTimeoutHandler(10))                            .addHandlerLast(new WriteTimeoutHandler(10))                                  )                             );         ClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);          return WebClient.builder()            .baseUrl("http://localhost:8080/findAll")            .clientConnector(connector)            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)            .build();}

spring.codec.max-in-memory-size=1MB

<dependency>    <groupId>co.paralleluniverse</groupId>    <artifactId>quasar-core</artifactId>    <version>0.7.4</version>    <classifier>jdk8</classifier></dependency>

-javaagent:path-to-quasar-jar.jar

<plugin>    <groupId>org.apache.maven.plugins</groupId>    <artifactId>maven-surefire-plugin</artifactId>    <configuration>        <!-- Turn off before production -->        <!-- argLine>-Dco.paralleluniverse.fibers.verifyInstrumentation=true</argLine -->
        <!-- Enable if using compile-time (AoT) instrumentation -->        <!-- argLine>-Dco.paralleluniverse.fibers.disableAgentWarning</argLine -->
        <!-- Quasar Agent for JDK 8 -->        <argLine>-javaagent:${co.paralleluniverse:quasar-core:jar:jdk8}</argLine>    </configuration></plugin>

new Fiber<V>() {    @Override    protected V run() throws SuspendExecution, InterruptedException {        // your code         } }.start();

new Fiber<Void>(new SuspendableRunnable() {    @Override    public void run() throws SuspendExecution, InterruptedException {        // your code    } }).start();

new Fiber<Void>((SuspendableRunnable) () -> {    // your code    }).start();

@Bean(name = EXECUTOR)public Executor setThreadPoolExecutor() {
    /* Do not drop task */    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(getCorePoolSize());    threadPoolTaskExecutor.setMaxPoolSize(getMaxPoolSize());    threadPoolTaskExecutor.setThreadNamePrefix("Executor_");    threadPoolTaskExecutor.setQueueCapacity(200);    threadPoolTaskExecutor.setKeepAliveSeconds(60);    threadPoolTaskExecutor.setAllowCoreThreadTimeOut(false);    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());    threadPoolTaskExecutor.initialize();
    // return threadPoolTaskExecutor;    return new FiberExecutorScheduler("Fiber_", threadPoolTaskExecutor);}