波多野结衣 蜜桃视频,国产在线精品露脸ponn,a v麻豆成人,AV在线免费小电影

本文介紹了如何使用電抗器3.x將LIST&lt；T&>；轉(zhuǎn)換為通量&lt；T&>的處理方法，對(duì)大家解決問題具有一定的參考價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧！

問題描述

我有一個(gè)Asyn Call Thrift接口：

public CompletableFuture<List<Long>> getFavourites(Long userId){
    CompletableFuture<List<Long>> future = new CompletableFuture();
    OctoThriftCallback callback = new OctoThriftCallback(thriftExecutor);
    callback.addObserver(new OctoObserver() {
        @Override
        public void onSuccess(Object o) {
            future.complete((List<Long>) o);
        }

        @Override
        public void onFailure(Throwable throwable) {
            future.completeExceptionally(throwable);
        }
    });
    try {
        recommendAsyncService.getFavorites(userId, callback);
    } catch (TException e) {
        log.error("OctoCall RecommendAsyncService.getFavorites", e);
    }
    return future;
}

現(xiàn)在它返回CompletableFuture&lt；列表&gt；。然后我調(diào)用它來使用Flux做一些處理器。

public Flux<Product> getRecommend(Long userId) throws InterruptedException, ExecutionException, TimeoutException {
    // do not like it
    List<Long> recommendList = wrapper.getRecommend(userId).get(2, TimeUnit.SECONDS);

    System.out.println(recommendList);
    return Flux.fromIterable(recommendList)
            .flatMap(id -> Mono.defer(() -> Mono.just(Product.builder()
                    .userId(userId)
                    .productId(id)
                    .productType((int) (Math.random()*100))
                    .build())))
            .take(5)
            .publishOn(mdpScheduler);
}

但是，我想從getFavourites方法中獲取一個(gè)通量，并且可以在getRecommend方法中使用它。
或者，您可以推薦Flux API，我可以將List<Long> recommendList轉(zhuǎn)換為Flux<Long> recommendFlux。

推薦答案

要將CompletableFuture<List<T>>轉(zhuǎn)換為Flux<T>，可以使用Mono#fromFuture和Mono#flatMapMany：

var future = new CompletableFuture<List<Long>>();
future.completeAsync(() -> List.of(1L, 2L, 3L, 4L, 5L),
    CompletableFuture.delayedExecutor(3, TimeUnit.SECONDS));

Flux<Long> flux = Mono.fromFuture(future).flatMapMany(Flux::fromIterable);

flux.subscribe(System.out::println);

List<T>在回調(diào)中異步接收到的Flux<T>也可以不使用CompletableFuture轉(zhuǎn)換為Flux<T>。
您可以直接使用Mono#create和Mono#flatMapMany：

Flux<Long> flux = Mono.<List<Long>>create(sink -> {
  Callback<List<Long>> callback = new Callback<List<Long>>() {
    @Override
    public void onResult(List<Long> list) {
      sink.success(list);
    }

    @Override
    public void onError(Exception e) {
      sink.error(e);
    }
  };
  client.call("query", callback);
}).flatMapMany(Flux::fromIterable);

flux.subscribe(System.out::println);

或簡單使用Flux#create一次多次排放：

Flux<Long> flux = Flux.create(sink -> {
  Callback<List<Long>> callback = new Callback<List<Long>>() {
    @Override
    public void onResult(List<Long> list) {
      list.forEach(sink::next);
    }

    @Override
    public void onError(Exception e) {
      sink.error(e);
    }
  };
  client.call("query", callback);
});

flux.subscribe(System.out::println);

這篇關(guān)于如何使用電抗器3.x將LIST&lt；T&>；轉(zhuǎn)換為通量&lt；T&>的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，