Возврат CompletableFuture без раскрытия потока исполнителя - PullRequest
Новая
       8

Возврат CompletableFuture без раскрытия потока исполнителя

2 голосов
/ 09 июня 2019

Я предоставляю метод в библиотеке, которая возвращает CompletableFuture. Вычисление этого метода происходит на однопоточном Executor, который является моим узким местом, поэтому я не хочу, чтобы какая-либо последующая работа выполнялась в том же потоке.

Если я воспользуюсь простым подходом возврата результата «supplyAsync», я буду показывать свой драгоценный поток вызывающим, которые могут добавлять синхронные операции (например, через thenAccept), которые могут занять некоторое время процессора в этом потоке.

Repro ниже: 

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CfPlayground {
    private ExecutorService preciousExecService = Executors.newFixedThreadPool(1);

    CfPlayground() {}

    private static void log(String msg) {
        System.out.println("[" + Thread.currentThread().getName() + "] " + msg);
    }

    CompletableFuture<String> asyncOp(String param) {
        return CompletableFuture.supplyAsync(() -> {
            log("In asyncOp");
            return "Hello " + param;
        }, preciousExecService);
    }

    void syncOp(String salutation) {
        log("In syncOp: " + salutation);
    }

    void run() {
        log("run");
        asyncOp("world").thenAccept(this::syncOp);
    }

    public static void main(String[] args) throws InterruptedException {
        CfPlayground compFuture = new CfPlayground();
        compFuture.run();
        Thread.sleep(500);
        compFuture.preciousExecService.shutdown();
    }
}

Это действительно печатает: 

[main] run
[pool-1-thread-1] In asyncOp
[pool-1-thread-1] In syncOp: Hello world

Одно решение, которое я нашел, состояло в том, чтобы представить другого Executor и добавить no-op thenApplyAsync с этим исполнителем перед возвратом CompletableFuture 

    CompletableFuture<String> asyncOp(String param) {
        return CompletableFuture.supplyAsync(() -> {
            log("In asyncOp");
            return "Hello " + param;
        }, preciousExecService).thenApplyAsync(s -> s, secondExecService);
    }

Это работает, но не выглядит супер элегантно - есть ли лучший способ сделать это?

Ответы [ 2 ]

0 голосов
/ 11 июня 2019

Нет возможности отсоединить ваше завершение от выполнения зависимого действия.Когда поток, соединяющий в цепочку зависимое действие, уже завершил регистрацию, и поток вашего исполнителя завершает будущее, какой другой поток должен выполнить зависимое действие, если не было назначено никакого другого исполнителя?

Ваш подход объединения цепочки с другим действиемдругой исполнитель, кажется, лучшее, что вы можете получить.Однако важно отметить, что в случае исключительного завершения исключение распространяется без оценки функций, переданных в thenApply.Распространение этого исключения может снова привести к открытию потока, если вызывающая сторона приковала действие, подобное whenComplete, handle или exceptionally.

С другой стороны, вам не нужноукажите вторичного исполнителя, так как вы можете использовать метод async без параметра executor, чтобы получить пул по умолчанию (общий Fork / Join).

Поэтому создание цепочки .whenCompleteAsync((x,y) -> {}) - лучшее решение дляВаша проблема до сих пор.

0 голосов
/ 10 июня 2019

Вы можете просто изменить сигнатуру метода, чтобы она возвращала Future вместо CompletableFuture: 

Future<String> asyncOp(String param) {
    return CompletableFuture.supplyAsync(() -> {
        log("In asyncOp");
        return "Hello " + param;
    }, preciousExecService);
}

Таким образом, метод run() выдаст ошибку компиляции: 

void run() {
    log("run");
    asyncOp("world").thenAccept(this::syncOp);
}

Вызывающий все равно сможет привести возвращенный Future обратно к CompletableFuture, но это будет довольно неправильным использованием вашего API, и это не может произойти случайно.

...