标签: completable-future

我是 Java 8 中 lambdas 和异步代码的新手。我不断得到一些奇怪的结果......

我有以下代码：

import java.util.concurrent.CompletableFuture;

public class Program {

    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            String test = "Test_" + i;
            final int a = i;

            CompletableFuture<Boolean> cf = CompletableFuture.supplyAsync(() -> doPost(test));
            cf.thenRun(() -> System.out.println(a)) ;
        }
    }

    private static boolean doPost(String t) {
        System.out.println(t);

        return true;
    }
}

实际代码要长得多，因为该doPost方法会将一些数据发布到 Web 服务。但是，我可以用这个简单的代码复制我的问题。

我想让该doPost方法执行 100 次，但出于性能原因异步执行（为了将数据推送到 Web 服务的速度比执行 100 次同步调用更快）。

在上面的代码中，“doPost”方法运行了随机次数，但始终不超过 20-25 次。没有抛出异常。似乎某些线程处理机制正在默默地拒绝创建新线程并执行它们的代码，或者线程在不使程序崩溃的情况下默默地崩溃。 …

我正在学习CompletableFuture,在下面的代码片段中,该thenAccept()方法不打印应该的值10,但程序编译无一例外.任何人都可以解释我的问题是什么？

import java.util.concurrent.*;
import java.util.stream.Stream;

public class CompletableFutureTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture.supplyAsync(CompletableFutureTest::counting).thenAccept(System.out::println);
        System.out.println("xD");
    }
    public static int counting() {


        Stream.iterate(1, integer -> integer +1).limit(5).forEach(System.out::println);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 10;
    }

}

我正在学习CompletableFutures.

我不是在问和之间的区别thenApply()thenCompose().相反,我想问一下感觉不对的代码"气味",以及实际可能证明它的理由.

从CompletableFuture我到目前为止看到的用法来看,似乎你从来没有这样:

CompletableFuture<String> foo = getSomething().thenApply((result) -> { ... });

这不是:

String foo = getSomething().thenCompose((result) -> { ... });

要返回未来,您必须使用thenCompose(),否则thenApply().

然而,从经验来看,语言并没有设法消除每次都做出这种明确的选择,这似乎很奇怪.例如,是否有一个单一的方法,thenDo()其返回类型是从returnlambda中推断出来的(在编译期间)？它可能然后被给予thenApply或thenCompose在编译时般的性能以及.

但我确信有一个很好的理由采用不同的方法,所以我想知道原因.

• 是因为在Java中从lambda推断返回类型是危险还是不可能？(我也是Java的新手.)

• 是不是因为有一种情况,单个方法确实不明确,唯一的解决方案是使用单独的方法？(我想象可能,嵌套CompletableFuture或复杂的界面和泛型.)如果是这样,有人可以提供一个明确的例子吗？

• 是出于其他原因还是有记录的推荐？

我试图包装CompletableFuture在 ReactorMono类型中以简化我的转换操作。Project Reactor 总体来说更方便！我在 AWS Lambda 函数中工作，我正在使用新的 AWS Java SDK 2.x 版本调用 AWS 服务，例如 S3、SQS 等。这个新的开发工具包允许对 AWS 服务进行异步调用并返回 CompleteableFuture 对象。

例如：

S3AsyncClient s3AsyncClient = S3AsyncClient.builder().build();    
Mono.fromFuture(s3AsyncClient.getObject(b -> 
b.bucket(bucketId).key(objectKey), AsyncResponseTransformer.toBytes()).subscribe()
System.out.println("stuff");

问题是，当我的主代码调用 时CompletableFuture (s3AsyncClient.getObject)，执行线程突然切换到 CompleteableFuture 线程并且我调用 Mono 的主方法在 CompletableFuture 完成之前返回。

基本上，从上面的例子来看，"stuff"字符串在s3AsyncClient.getObject完成之前被打印出来。

如何确保 Mono 和CompletableFuture在同一线程中执行，或者如何确保我的 lambda 在CompletableFuture完成之前不会终止？

对于那些想知道的人，只有在我将代码远程部署到 AWS Lambda 时才会出现这种行为。我在本地没有遇到这种行为......

如何避免单元测试中的手动睡眠。假设在下面的代码中，Process和notify需要大约 5 秒的时间进行处理。所以为了完成处理，我添加了 5 秒的睡眠。

public class ClassToTest {

    public ProcessService processService;
    public NotificationService notificationService;

    public ClassToTest(ProcessService pService ,NotificationService nService ) {
        this.notificationService=nService;
        this.processService = pService;
    }
    public CompletableFuture<Void> testMethod()
    {
          return CompletableFuture.supplyAsync(processService::process)
                        .thenAccept(notificationService::notify);
    }

}

有没有更好的方法来处理这个问题？

 @Test
    public void comletableFutureThenAccept() {
         CompletableFuture<Void> thenAccept = 
          sleep(6);
          assertTrue(thenAccept.isDone());  
          verify(mocknotificationService, times(1)).notify(Mockito.anystring());
    }

我正在使用 MDC Logger，除了一种情况外，它对我来说非常有用。无论我们在代码中的何处使用 CompletableFuture，对于创建的线程，MDC 数据都不会传递到下一个线程，因此日志失败。例如，在我使用以下代码段创建新线程的代码中。

CompletableFuture.runAsync(() -> getAcountDetails(user));

日志结果如下

2019-04-29 11:44:13,690 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] RestServiceExecutor:  service: 
2019-04-29 11:44:13,690 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] RestServiceExecutor: 
2019-04-29 11:44:13,779 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] UserDetailsRepoImpl: 
2019-04-29 11:44:13,950 INFO   [ForkJoinPool.commonPool-worker-3] RestServiceExecutor:  header: 
2019-04-29 11:44:13,950 INFO   [ForkJoinPool.commonPool-worker-3] RestServiceExecutor:  service: 
2019-04-29 11:44:14,012 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieving Config Data details.
2019-04-29 11:44:14,028 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieved Config Data details : 1
2019-04-29 11:44:14,028 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieving Config Data …

我正在学习 Java 8 以及更详细的“CompletableFuture”。遵循这个有趣的教程： https://www.callicoder.com/java-8-completablefuture-tutorial/

我编写了以下 Java 类：

package parallels;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import javax.ws.rs.client.ClientRequestFilter;
import javax.ws.rs.core.Response;

import org.jboss.resteasy.client.jaxrs.ResteasyClient;
import org.jboss.resteasy.client.jaxrs.ResteasyClientBuilder;
import org.jboss.resteasy.client.jaxrs.ResteasyWebTarget;




public class Test {
    private static final String USER_AGENT = "Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:56.0) Gecko/20100101 Firefox/56.0";
    private static final Executor executor = Executors.newFixedThreadPool(100);

    public static void main(String[] args) {

        List<String> webPageLinks= new ArrayList<String>();
        for (int i=0;i<30;i++) {
            webPageLinks.add("http://jsonplaceholder.typicode.com/todos/1");
        }

        // Download contents of all the …

我正在尝试维护多个线程中的项目列表，每个线程一个（例如，每个线程说一个套接字连接）。我在维护此列表ArrayDeque<>。我面临的问题ArrayDeque<>是超过项目数超过项目数。线程池中的线程数。

这是我的代码：

package com.practice;

import java.util.ArrayDeque;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CompletableFutureApp implements AutoCloseable {
    private static void sleep(long timeMS) {
        try {
            Thread.sleep(timeMS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try (CompletableFutureApp completableFutureApp = new CompletableFutureApp()) {

            Runnable[] tasksList1 = new Runnable[100];

            for (int i = 0; i < tasksList1.length; i++) {
                String msg1 = "TaskList 1 no.: " + i; …

有人可以帮助为下面的代码部分（尤其是添加回调部分）编写 JUnit 测试吗？我不知道如何使用回调为 Listenablefuture 编写单元测试用例

    private void handleResponse(final ListenableFuture<UserRecordResult> response, CompletableFuture future) {

        Futures.addCallback(response, new FutureCallback<UserRecordResult>() {

            @Override
            public void onFailure(@Nonnull Throwable throwable) {

                future.completeExceptionally(new Exception("Fail to put record" + throwable.getMessage()));
            }

            @Override
            public void onSuccess(UserRecordResult result) {
                if(result.isSuccessful()) {
                    future.complete(true);
                } else {
                    future.completeExceptionally(new Exception("Fail to put record"));
                }
            }
        });
    }

在下面的代码中，无论我将 i 的最大值设置为多少，线程总数都不会超过 13。它使用的是什么线程池？我在哪里可以找到它的默认设置？

public static void main(String[] args) {
    // write your code here
    for (int i = 0; i <= 5; i++) {

        System.out.println("kick off" + i);
        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);

                System.out.println(java.lang.Thread.activeCount());
            }
            catch (Exception e) {
                System.out.println("error");
            }
        });

    }
    System.out.println(java.lang.Thread.activeCount());
    try {
        Thread.sleep(10000);
    }
    catch (InterruptedException e) {
        e.printStackTrace();
    }
}