标签: completable-future

我创建了一个示例,我想知道如何使用CompletableFuture？返回值？我也改变了CompletableFuture<Void> exeFutureList,CompletableFuture<Integer> exeFutureList但是eclipse总是建议把它设置回Void.

请告诉我如何使用CompletableFuture返回值.

代码:

    public class MainClass {

    static ExecutorService exe = null;
    static CompletableFuture<Void> exeFutureList = null;

    public static void main(String[] args) {
        exe = Executors.newFixedThreadPool(1);
        exeFutureList = CompletableFuture.runAsync(new RunClass(8), exe);
    }

    static class RunClass implements Runnable {

        private int num;

        public RunClass(int num) {
            // TODO Auto-generated constructor stub
            this.num = num;
        }

        public void run() {
            // TODO Auto-generated method stub
            this.num = this.num + 10;
        }

    }
}

我一直在使用CompletableFuture.allOf(...)帮助程序来创建汇总期货，这些期货只有在其复合期货标记为已完成时才会变为“完成”，即：

CompletableFuture<?> future1 = new CompletableFuture<>();
CompletableFuture<?> future2 = new CompletableFuture<>();
CompletableFuture<?> future3 = new CompletableFuture<>();

CompletableFuture<?> future = CompletableFuture.allOf(future1, future2, future3);

我希望对此功能稍作更改，在以下情况下，总的未来市场是完整的：

• 所有期货均已成功完成或
• 任何一个未来都没有成功完成

在后一种情况下，合计期货应立即（例外）完成，而不必等待其他期货完成（即快速失败）。

为了说明这一点，请CompletableFuture.allOf(...)考虑一下：

// First future completed, gotta wait for the rest of them...
future1.complete(null);
System.out.println("Future1 Complete, aggregate status: " + future.isDone());

// Second feature was erroneous! I'd like the aggregate to now be completed with failure
future2.completeExceptionally(new Exception());
System.out.println("Future2 Complete, aggregate status: " + future.isDone());

// Finally complete …

我正在尝试.exceptionally和.handle,但那些似乎不起作用.在scala中,你可以使用一个类似于finally块的闭包来调用未来的方法(它在异常AND成功时运行)并且它按原样传播异常或链上的成功.

我试过这个......

    CompletableFuture<Object> future = newFuture.handle((r, e) -> {
        if(r != null)
            return r;
        else if(e != null)
            return e;
        else
            return new RuntimeException("Asdf");            
    });

    Assert.assertTrue(future.isCompletedExceptionally());

但是测试失败了,因为未来完全成功,异常的结果(多么奇怪).

我刚刚阅读了有关的文档，CompletableFuture::runAsync并对此解释感到困惑。这是那里写的：

返回一个新代码CompletableFuture，该新代码由在给定执行程序中运行的任务运行给定操作后异步完成。

据我了解，它CompletableFuture看起来Future可以“注册”某种回调并在给定操作完成后隐式调用它们。

考虑到这一点，让我们考虑以下代码：

ExecutorService threadsPool;
Runnable r;
//...
CompletableFuture.runAsync(r, threadsPool);

在此代码中，我们在中注册了Runnable要异步执行的ThreadPool。

但是CompletableFuture，由任务异步完成意味着什么。任务如何使CompletableFuture完成...？这对我来说没有多大意义。

我刚刚开始研究Java的CompletableFuture,并对这是否真正异步(即同时在一个线程上运行)或跨越多个线程(并行)有点困惑.

例如,假设我想要进行1000次不同的服务呼叫.进一步假设每个服务调用可以异步进行.使用CompletableFuture时,JVM是否会生成1000个单独的线程(假设JVM允许这么多线程),或者在一个线程中执行所有这些请求？或者是两者兼而有之？使用一些线程异步执行这些请求？

我想做的是这样的(在Python中):https: //pawelmhm.github.io/asyncio/python/aiohttp/2016/04/22/asyncio-aiohttp.html

有没有办法在Java中异步执行同一个线程上的多个请求/操作？

执行async时CompletableFuture，父线程org.slf4j.MDC上下文以及上下文会丢失。

这很不好，因为我使用某种“鱼标签”来跟踪多个日志文件中一个请求的日志。

MDC.put("fishid", randomId())

问题：一般情况下，我该如何保留该ID CompletableFutures？

List<CompletableFuture<UpdateHotelAllotmentsRsp>> futures =
    tasks.stream()
        .map(task -> CompletableFuture.supplyAsync(
            () -> businesslogic(task))
        .collect(Collectors.toList());

List results = futures.stream()
    .map(CompletableFuture::join)
    .collect(Collectors.toList());

public void businesslogic(Task task) {
       LOGGER.info("mdc fishtag context is lost here");
}

我试图在某些数据库异常上使用重试策略实现数据库查询.重试策略的代码不是很相关,所以我没有包含它.正如您在下面的代码中看到的那样 - 我编写了一个retryCallable,它采用了重试策略和Callable populateData().

在getDataFromDB,我从DB获取数据并将数据放在全局散列图中,该散列图在应用程序级别充当缓存.

此代码按预期工作.我想populateData从另一个班级调用.但是,这将是一个阻止呼叫.由于这是数据库并且具有重试策略,因此这可能很慢.我想populateData异步调用.

我如何使用CompletableFuture或FutureTask来实现这一目标？ CompletableFuture.runAsync期待一个可运行的.CompletableFuture.supplyAsync期待供应商.我以前没有实现过这些东西.所以关于最佳实践的任何建议都会有所帮助.

Class TestCallableRetry {

public void populateData() {
        final Callable<Set<String>> retryCallable = new RetryingCallable<>(retryStrategyToRetryOnDBException(), getDataFromDB());
        Set<String> data = new HashSet<>();

        data = retryCallable.call();

        if (data != null && !data.isEmpty()) {
            // store data in a global hash map
        }
    }

    private Callable<Set<Building>> getDataFromDB() {
        return new Callable<Set<String>>() {
            @Override
            public Set<String> call() {
                // returns data from database
            }
        };
    }
}

Class …

我已经在这几个小时,但似乎无法解开这个.该错误与此段代码有关:

(这个问题底部的完整代码.我正在使用屏幕截图直观地显示问题.)

错误本身是:

没有类型变量的实例U存在,以便GetUsersForAdAccountResponse符合CompletionStage

在thenComposes 链的最开始,我擦除return并使用IntelliJ的"引入局部变量..."功能来查看整个链(直到并包括第1118行)返回的类型:

结果是

final CompletionStage<U> uCompletionStage = ...

但是你可以看到包含方法的返回类型是

public CompletionStage<GetUsersForAdAccountResponse> ...

什么阻止编译器推断GetUsersForAdAccountResponse？(再次,通常这里有一个return.)

我还尝试在每个过程中引入局部变量thenCompose,它们看起来都是正确的.每个产生一个CompletionStage<Foo>,下thenCompose一个Foo产生一个lambda期望并产生一个CompletionStage<Bar>,等等.(在一次代码重组中,我确实看到了嵌套,CompletionStage<CompletionStage<Foo>>但我认为这是我自己重写的一件神器.)

我不知道它是否会有所帮助,但这是整个方法:

  @Override
  public CompletionStage<GetUsersForAdAccountResponse> getUsersForAdAccount(
      RequestContext context, GetUsersForAdAccountRequest request) {

    Uuid adAccountId = request.getAdAccountId();

    return verifyAuthorization(context,
        PortcullisTemplates.Action.GET_USERS_FOR_AD_ACCOUNT.getName(),
        portcullisTemplates.topOrganizationResource())
        .thenCompose(auditLogPrincipal -> jdbiExecutor.executeInTransaction(handler -> {

              // We purposely safeguard the account lookup as well behind Portcullis.
              AdAccountDao adAccountDao = handler.attach(AdAccountDao.class);
              if (adAccountDao.getAdAccountById(adAccountId) …

我有一个代码:

CompletableFuture<Integer> c1 = new CompletableFuture<Integer>()
        .thenApply((data) -> data * 2);
c1.thenAccept(System.out::println);

c1.complete(20);


CompletableFuture<Integer> c2 = new CompletableFuture<>();

c2.thenApply(data -> data * 2)
        .thenAccept(System.out::println);
c2.complete(20);

输出:

20 40

题:

1. 为什么c1和c2之间的输出有差异？
2. 为什么需要通过调用来重复c1中的未来类型:

新的CompletableFuture < 整数 >()

我想并行执行多个数据库查询，并将结果存储在映射中。我正在尝试这样做，但是访问地图时地图没有完全填充。

我做错什么了吗？

 public Map<MapKeyEnums, Set<String>> doDBCalls(String phoneNumber, long timestamp) {

         Map<MapKeyEnums, Set<String>> instrumentsEdgesMap = new EnumMap<>(MapKeyEnums.class);

         CompletableFuture.supplyAsync(() -> dbReadService.getCall(phoneNumber, PhoneNumber.class, "ABC", timestamp)).
                    thenApply(x -> instrumentsEdgesMap.put(MapKeyEnums.ABC, x));

         CompletableFuture.supplyAsync(() -> dbReadService.getCall(phoneNumber, PhoneNumber.class, "XYZ", timestamp)).
                    thenApply(x -> instrumentsEdgesMap.put(MapKeyEnums.XYZ, x));

         CompletableFuture.supplyAsync(() -> dbReadService.getCall(phoneNumber, PhoneNumber.class, "DEF", timestamp)).
                    thenApply(x -> instrumentsEdgesMap.put(MapKeyEnums.DEF, x));

         return instrumentsEdgesMap;

}

任何帮助将不胜感激，在此先感谢。