Выполнять вычисления IO параллельно в Java8

Я знаком с функциональными языками программирования, обычно в Scala и Javascript. Я работаю над проектом Java8 и не знаю, как я должен проходить через список/поток элемента, а также выполнять побочный эффект для каждого из них параллельно, используя собственный пул потоков и возвращать объект, на котором его можно послушать для завершения (пусть успех или неудача).

В настоящее время у меня есть следующий код, он работает (я использую реализацию Promise Promise в качестве возврата), но это кажется не идеальным, потому что ForkJoinPool не предназначен для использования для интенсивных вычислений ввода-вывода в первую очередь.

public static F.Promise<Void> performAllItemsBackup(Stream<Item> items) {
    ForkJoinPool pool = new ForkJoinPool(3);
    ForkJoinTask<F.Promise<Void>> result = pool
            .submit(() -> {
                try {
                    items.parallel().forEach(performSingleItemBackup);
                    return F.Promise.<Void>pure(null);
                } catch (Exception e) {
                    return F.Promise.<Void>throwing(e);
                }
            });

    try {
        return result.get();
    } catch (Exception e) {
        throw new RuntimeException("Unable to get result", e);
    }
}

Может ли кто-нибудь дать мне более идиоматическую реализацию вышеуказанной функции? В идеале, не используя ForkJoinPool, используя более стандартный тип возврата и самые последние API-интерфейсы Java8? Не уверен, что я должен использовать между CompletableFuture, CompletionStage, ForkJoinTask...

Ответы

Ответ 1

Каноническое решение было бы

public static CompletableFuture<Void> performAllItemsBackup(Stream<Item> items) {
    ForkJoinPool pool = new ForkJoinPool(3);
    try {
        return CompletableFuture.allOf(
            items.map(CompletableFuture::completedFuture)
                 .map(f -> f.thenAcceptAsync(performSingleItemBackup, pool))
                 .toArray(CompletableFuture<?>[]::new));
    } finally {
        pool.shutdown();
    }
}

Обратите внимание, что взаимодействие между пулом ForkJoin и параллельными потоками - это неопределенная реализация, на которую вы не должны полагаться. Напротив, CompletableFuture предоставляет выделенный API для обеспечения Executor. Это даже не должно быть ForkJoinPool:

public static CompletableFuture<Void> performAllItemsBackup(Stream<Item> items) {
    ExecutorService pool = Executors.newFixedThreadPool(3);
    try {
        return CompletableFuture.allOf(
            items.map(CompletableFuture::completedFuture)
                 .map(f -> f.thenAcceptAsync(performSingleItemBackup, pool))
                 .toArray(CompletableFuture<?>[]::new));
    } finally {
        pool.shutdown();
    }
}

В любом случае вы должны явно отключить исполнителя, вместо того чтобы полагаться на автоматическую очистку.

Если вам нужен результат F.Promise<Void>, вы можете использовать

public static F.Promise<Void> performAllItemsBackup(Stream<Item> items) {
    ExecutorService pool = Executors.newFixedThreadPool(3);
    try {
        return CompletableFuture.allOf(
            items.map(CompletableFuture::completedFuture)
                 .map(f -> f.thenAcceptAsync(performSingleItemBackup, pool))
                 .toArray(CompletableFuture<?>[]::new))
            .handle((v, e) -> e!=null? F.Promise.<Void>throwing(e): F.Promise.pure(v))
            .join();
    } finally {
        pool.shutdown();
    }
}

Но обратите внимание, что это, как и ваш исходный код, возвращается только после завершения операции, а методы, возвращающие CompletableFuture, позволяют выполнять операции асинхронно до тех пор, пока вызывающий абонент не вызовет join или get.

Чтобы вернуть истинно асинхронный Promise, вы должны обернуть всю операцию, например.

public static F.Promise<Void> performAllItemsBackup(Stream<Item> stream) {
    return F.Promise.pure(stream).flatMap(items -> {
        ExecutorService pool = Executors.newFixedThreadPool(3);
        try {
            return CompletableFuture.allOf(
                items.map(CompletableFuture::completedFuture)
                     .map(f -> f.thenAcceptAsync(performSingleItemBackup, pool))
                     .toArray(CompletableFuture<?>[]::new))
                .handle((v, e) -> e!=null? F.Promise.<Void>throwing(e): F.Promise.pure(v))
                .join();
        } finally {
            pool.shutdown();
        }
    });
}

Но лучше выбрать один API вместо того, чтобы перепрыгивать назад и вперед между двумя различными API.