标签: completable-future

我一直在使用 Java 的 CompletableFuture 这样

CompletableFuture.runAsync(() -> {//Some code here });

当我尝试在此代码块中使用 Resteasy Client 时，我得到一个

javax.ws.rs.ProcessingException: Unable to find a MessageBodyReader of content-type application/json;charset=utf-8 and type class java.lang.String

如果我在 completablefuture 之外使用客户端，则它可以工作。Resteasy 代码看起来像这样

        ResteasyClient client = new ResteasyClientBuilder().build();
        client.register(new AcceptEncodingFilter("gzip"));
        ResteasyWebTarget target = client.target(exampleURL);

        target = target.queryParam("1", 1)
                .queryParam("2", "1")
                .queryParam("3", 3)
                .queryParam("4", 4)
                .queryParam("5", "5");

        Response response = target.request().get();
        resultString = response.readEntity(String.class);

我将在 completablefuture 之外运行 resteasy 代码来“修复”问题，但想了解为什么会发生这种情况。

CompletableFuture 中的 resteasy 代码如下所示：

CompletableFuture.runAsync(() -> {
            try {
                ResteasyClient client = new …

我正在玩弄 Java8 的流和CompletableFutures。我预先存在的代码有一个类，它接受一个 URL 并下载它：

public class FileDownloader implements Runnable {
    private URL target;
    public FileDownloader(String target) {
        this.target = new URL(target);
    }
    public void run() { /* do it */ }
}

现在，这个类从另一个发出的部分List<String>（单个主机上的多个目标）获取它的信息。

我已将周围的代码切换为CompletableFuture：

public class Downloader {
    public static void main(String[] args) {
        List<String> hosts = fetchTargetHosts();
        for (String host : hosts) {
            HostDownloader worker = new HostDownloader(host);
            CompletableFuture<List<String>> future = 
                CompletableFuture.supplyAsync(worker);
            future.thenAcceptAsync((files) -> {
                for (String target : files) {
                    new …

该CompletableFutureAPI是相当吓人的，接受的大量的，和thens和其他的东西; 很难说为什么存在不同的选择。

CompletableFuture<?> future = CompletableFuture.supplyAsync(() ->..., executor)

future.startNonBlocking...( (...) -> { callback behavior done when complete }

我基本上是在尝试模仿一个new Thread(() -> dostuff).start()但具有更好的线程池、错误处理等功能。 注意：我实际上并不需要Runnable这里的接口，我正在生成一段现有代码。

启动异步任务并在完成后执行行为的正确方法是什么？或处理抛出的异常？

你好，我认为使用 CompletableFuture 和默认设置ForkJoinPool我可以优化任务的执行而不是经典，ExecutorService但我错过了一些东西

使用此代码执行需要 1 秒，我有 3 个工作线程：

for (int i = 0; i < 3; i++) {
    final int counter = i;
    listTasks.add(CompletableFuture.supplyAsync(() -> {
        Thread.sleep(1000);
        System.out.println("Looking up " + counter + " on thread " + Thread.currentThread().getName());
        return null;
    }));
}

好吧，看起来很正常。

但是使用此代码，需要 3 秒：

for (int i = 0; i < 9; i++) {
    final int counter = i;
    listTasks.add(CompletableFuture.supplyAsync(() -> {
        Thread.sleep(1000);
        System.out.println("Looking up " + counter + " on thread " …

我有一些误解的合同的thenCompose(Function<? super T,? extends CompletionStage<U>> fn)。这是那里所说的：

返回一个新的CompletionStage，当这个阶段正常完成时，将这个阶段作为提供函数的参数执行。

该函数似乎接受 this 的结果CompletionStage，而不是它CompletionStage本身。那么他们是什么意思呢？

那么返回的 代表的任务CompletableFuture呢？

Runnable r;
ExecutorService es;
Function<Void, CompletableFuture<Void>>f;
//...
CompletableFuture.runAsync(r, es)
.thenCompose(f);

这是否意味着，由CompletableFuture返回的任务thenCompose将ThreadPool与Runnable r?

在 java 中使用 CompletableFuture 时遇到问题。我有 2 个选择请求，它们是在从服务器接收响应时填充的。

在连接线程（THREAD-1）（使用反应器）中，我使用：

if(hasException) {
   selectFuture.completeExceptionally(new ClientException(errorCode));
} else {
   System.out.println("Before complete future");
   selectFuture.complete(result);
   System.out.println("After complete future");
}

在其他线程（THREAD-2）中，我使用：

CompleteFuture.allOf(allSelect).whenComplete((aVoid, throwable) -> {
   System.out.println("Receive all future");
   // Do sth here
});

我的情况是系统打印出“接收所有未来”但调用时 THREAD-1 被阻止future.complete(result);它无法退出该命令。如果在 THREAD-2 中，我使用CompletableFuture.allOf(allOfSelect).get()，THREAD-1 将正确运行。但是使用CompletableFuture.get()会降低性能，所以我想使用CompletableFuture.whenComplete().

谁能帮我解释一下阻塞的原因？

谢谢！

我正在尝试处理 Kotlin 中的 CompletableFuture 异常，但我无法弄清楚如何提供适当的参数。所以，例如，我有：

CompletableFuture.runAsync { "sr" } .exceptionally{e -> {}}

但随后编译器抱怨Cannot infer type parameter T。

我该如何解决？

exceptionally似乎不允许在 CompletionStage的方法内重新抛出异常。

我需要检查某种异常，如果没有，我需要重新抛出它：

Future<JsonNode> futureSite = someClient.getSite(siteId, queryParams);

CompletionStage<JsonNode> outcome = FutureConverters.toJava(futureSite);

return outcome.thenApplyAsync((siteJson) -> {
            Site site = Json.fromJson(siteJson, Site.class);
            try {
                return function.apply(site);
            } catch (RequestException e) {
                return e.result;
            }
        }, httpExecutionContext.current()).exceptionally(throwable -> {
            if(throwable instanceof SomeClientException) {
                if(((SomeClientException) throwable).httpStatusCode == 404) {
                   return entityNotFound("Site", siteId);
                }
            }


 // let JSON parsing failures and other errors bubble up, TODO play2.5


         throw throwable;
        });

throw throwable 错误说 unhandledException java.lang.Throwable

什么接口可能允许重新抛出异常？或者有更好的方法吗？

更新 ：

我尝试了以下 Holger 的建议，但我仍然不确定在什么时候我可以重新抛出这个：

BaseController.java : …

假设我们正在调用一个缓慢的远程服务来获得String响应。为此，我们调用了一个如下所示的方法：

public CompletableFuture<String> getSlowly(String param) {...}

在这种情况下，我们也知道我们知道该服务在我们调用它的 20% 的时间内不会返回有用的响应。因此，在我们调用的 20% 的情况下，getSlowly我们会得到一个异常完整的未来，并带有一些包装异常。意思是，我们将catch在以下调用的块中结束：

try {
    String response = getSlowly("some input").join();
    doSomething(response); // 80% succeed
catch (Exception e) {
    LOG.warn("Couldn't get response. Too bad."); // 20% fail
}

这个调用并不重要，我们的应用程序可以优雅地处理丢失的响应。但是，每当我们设法成功获得响应时，我们确实希望使用响应。

因为我们知道这种失败会经常发生，即我们确实预料到了，将这 20% 的调用视为例外是否有意义？

不将 20% 视为例外的一种方法是将 的实现更改为getSlowly以下内容：

public CompletableFuture<Optional<String>> getSlowly(String param) {...}

这看起来很笨拙。

有没有一种惯用的方式来处理这种脆弱的 CompletableFutures？当缺少返回值时，我们应该将其视为异常状态，还是应该返回可选项（或类似的自定义对象）？

我试图理解Java 8中的CompletableFuture.作为其中的一部分,我正在尝试进行一些REST调用以巩固我的理解.我正在使用这个库来进行REST调用:https://github.com/AsyncHttpClient/async-http-client.

请注意,此库返回GET调用的Response对象.

以下是我要做的事情:

1. 请调用此URL,该URL提供用户列表:https://jsonplaceholder.typicode.com/users
2. 使用GSON将响应转换为用户对象列表.
3. 迭代列表中的每个User对象,获取userID,然后从以下URL获取用户发布的帖子列表:https://jsonplaceholder.typicode.com/posts？userId = 1
4. 使用GSON将每个帖子响应转换为Post对象.

5. 构建UserPost对象的集合,每个对象都有一个用户对象和用户发布的帖子列表.

   public class UserPosts {
   
   private final User user;
   private final List<Post> posts;
   
   public UserPosts(User user, List<Post> posts) {
       this.user = user;
       this.posts = posts;
   }
   
   @Override
   public String toString() {
       return "user = " + this.user + " \n" + "post = " + posts+ " \n \n";
   }
   

   Run Code Online (Sandbox Code Playgroud)

   }

我目前实现如下:

package com.CompletableFuture;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture; …