引言

虽然2024年9月18号推出了JDK23,d但是目前项目中使用的大部分还是JDK8 (至少本公司的很多项目还在大量使用),在JDK8版本中，函数式接口是其一重大亮点。提升Java函数式编程的实用工具类在现代软件开发中，函数式编程已经成为了一种重要的编程范式，它强调使用函数来处理数据和构建逻辑。Java 8引入了函数式接口和Lambda表达式，极大地简化了代码的编写并提高了开发效率。本文将基于FunctionalUtils的工具类，展示一系列的实用功能，包括异步执行、事务处理等。

设计理念

FunctionalUtils的设计理念是基于Java函数式接口，如Supplier、Consumer、Function和Predicate，这些接口为实现项目需求时提供了一种强大的方式来传递行为，而不仅仅是数据。FunctionalUtils旨在提供一种简洁的方式来执行常见的函数式操作，同时增加了异步执行和事务处理等高级功能，以满足更复杂的业务场景。

核心功能

源码设计

package com.dereksmart.crawling.fuc;

import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.Collection;
import java.util.function.Predicate;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.*;
/**
 * @Author derek_smart
 * @Date 2024/9/20 8:08
 * @Description FunctionalUtils工具类
 */
public class FunctionalUtils {

    // 共享线程池，用于异步操作
    private static final ExecutorService executorService = Executors.newCachedThreadPool();

    /**
     * 从Supplier获取值，并返回它。
     */
    public static <T> T get(Supplier<T> supplier) {
        return supplier.get();
    }

    /**
     * 处理对象T，使用Consumer进行消费。
     */
    public static <T> void consume(T item, Consumer<T> consumer) {
        consumer.accept(item);
    }

    /**
     * 应用函数到输入值，并返回结果。
     */
    public static <T, R> R apply(T item, Function<T, R> function) {
        return function.apply(item);
    }

    /**
     * 重复执行Supplier提供的操作，直到满足条件。
     */
    public static <T> void repeatUntil(Supplier<Boolean> condition, Supplier<T> operation) {
        while (!condition.get()) {
            operation.get();
        }
    }

    /**
     * 对象转换，将输入对象T使用转换函数转换为R。
     */
    public static <T, R> R transform(T input, Function<T, R> transformer) {
        return transformer.apply(input);
    }

    /**
     * 安全地执行一个可能抛出异常的操作，异常被捕获并处理。
     */
    public static void runSafely(Runnable operation, Consumer<Exception> exceptionHandler) {
        try {
            operation.run();
        } catch (Exception e) {
            exceptionHandler.accept(e);
        }
    }

    /**
     * 创建并返回一个新对象，通过Supplier接口。
     */
    public static <T> T create(Supplier<T> supplier) {
        return supplier.get();
    }

    /**
     * 执行一个条件检查，如果条件为真，则执行操作。
     */
    public static void doIf(Supplier<Boolean> condition, Runnable action) {
        if (condition.get()) {
            action.run();
        }
    }

    /**
     * 执行一个操作，并返回一个状态值，通常用于链式操作中的状态检查。
     */
    public static boolean tryPerform(Runnable action) {
        try {
            action.run();
            return true;
        } catch (Exception e) {
            return false;
        }
    }


    /**
     * 对集合中的每个元素执行给定的操作。
     */
    public static <T> void forEach(Collection<T> collection, Consumer<T> action) {
        for (T item : collection) {
            action.accept(item);
        }
    }

    /**
     * 对集合中的元素进行过滤，并返回一个新的集合。
     */
    public static <T> Collection<T> filter(Collection<T> collection, Predicate<T> predicate) {
        Collection<T> result = createCollectionInstance(collection);
        for (T item : collection) {
            if (predicate.test(item)) {
                result.add(item);
            }
        }
        return result;
    }

    /**
     * 创建与给定集合类型相同的空集合实例。
     */
    private static <T> Collection<T> createCollectionInstance(Collection<T> collection) {
        try {
            return collection.getClass().newInstance();
        } catch (InstantiationException | IllegalAccessException e) {
            throw new RuntimeException("Could not create a new instance of the collection class", e);
        }
    }

    /**
     * 对集合中的元素应用转换函数，并返回一个新的集合。
     */
    public static <T, R> Collection<R> map(Collection<T> collection, Function<T, R> mapper) {
        Collection<R> result = (Collection<R>) createCollectionInstance(collection);
        for (T item : collection) {
            result.add(mapper.apply(item));
        }
        return result;
    }

    /**
     * 异步执行一个操作，返回CompletableFuture。
     */
    public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return CompletableFuture.runAsync(runnable, executorService);
    }

    /**
     * 异步执行一个有返回值的操作，返回CompletableFuture。
     */
    public static <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier) {
        return CompletableFuture.supplyAsync(supplier, executorService);
    }

    /**
     * 异步执行一个操作，并在操作完成时使用Consumer处理结果。
     */
    public static <T> void runAsync(T item, Consumer<T> consumer) {
        CompletableFuture.runAsync(() -> consumer.accept(item), executorService);
    }

    /**
     * 异步执行一个有返回值的操作，并应用Function处理结果。
     */
    public static <T, R> CompletableFuture<R> applyAsync(T item, Function<T, R> function) {
        return CompletableFuture.supplyAsync(() -> function.apply(item), executorService);
    }

    /**
     * 使用CompletableFuture执行一个操作，并在操作完成时进行回调。
     */
    public static <T> void whenCompleteAsync(Supplier<T> supplier, BiConsumer<? super T, ? super Throwable> action) {
        CompletableFuture.supplyAsync(supplier, executorService).whenCompleteAsync(action, executorService);
    }

    /**
     * 执行一个事务性操作，确保所有步骤都成功完成，否则回滚。
     */
    public static <T> boolean performTransaction(Supplier<Boolean>... operations) {
        boolean success = true;
        for (Supplier<Boolean> operation : operations) {
            if (!operation.get()) {
                success = false;
                break;
            }
        }
        if (!success) {
            // Rollback logic if necessary
        }
        return success;
    }


    /**
     * 关闭工具类使用的线程池资源，应当在应用程序关闭时调用。
     */
    public static void shutdownExecutorService() {
        executorService.shutdown();
    }
}

提供了以下功能：

• get(Supplier<T> supplier): 从Supplier获取值。
• consume(T item, Consumer<T> consumer): 对象消费，执行Consumer的操作。
• apply(T item, Function<T, R> function): 对象转换，将输入对象转换为另一种类型。
• repeatUntil(Supplier<Boolean> condition, Supplier<T> operation): 重复执行操作，直到条件满足。
• transform(T input, Function<T, R> transformer): 对象转换的另一种形式。
• runSafely(Runnable operation, Consumer<Exception> exceptionHandler): 安全执行操作，异常通过Consumer处理。
• create(Supplier<T> supplier): 创建对象实例。
• doIf(Supplier<Boolean> condition, Runnable action): 条件执行，如果条件为真，则执行操作。
• tryPerform(Runnable action): 尝试执行操作，返回执行成功或失败的布尔值。
• forEach(Collection<T> collection, Consumer<T> action): 遍历集合，对每个元素执行操作。
• filter(Collection<T> collection, Predicate<T> predicate): 过滤集合，根据条件返回满足条件的元素集合。
• map(Collection<T> collection, Function<T, R> mapper): 转换集合，对集合中的每个元素应用转换函数，返回转换后的元素集合。
• createCollectionInstance(Collection<T> collection): 创建与给定集合类型相同的空集合实例，用于filter和map方法。
• runAsync(Runnable runnable): 异步执行一个无返回值的操作。
• supplyAsync(Supplier<T> supplier): 异步执行一个有返回值的操作。
• runAsync(T item, Consumer<T> consumer): 异步执行一个消费者操作。
• applyAsync(T item, Function<T, R> function): 异步执行一个函数操作并返回结果。
• whenCompleteAsync(Supplier<T> supplier, BiConsumer<? super T, ? super Throwable> action): 异步执行一个操作并在操作完成时进行回调。
• performTransaction(Supplier<Boolean>... operations): 执行一个事务性操作，确保所有步骤都成功完成，否则执行回滚逻辑。
• shutdownExecutorService(): 关闭线程池资源，当应用程序关闭时应当调用此方法。

类图

实际应用

测试类

package com.dereksmart.crawling.fuc;


import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
/**
 * @Author derek_smart
 * @Date 2024/9/20 8:28
 * @Description FunctionalUtils测试类
 */
public class FunctionalUtilsExample {

    public static void main(String[] args) {
        // 异步执行无返回值的操作
        CompletableFuture<Void> future = FunctionalUtils.runAsync(() -> {
            performSomeLongRunningOperation();
            System.out.println("Async operation performed on thread: " + Thread.currentThread().getName());
        });

        // 异步执行有返回值的操作，并处理结果
        CompletableFuture<Integer> futureWithResult = FunctionalUtils.supplyAsync(() -> {
            System.out.println("Async operation with result performed on thread: " + Thread.currentThread().getName());
            return calculateSomeValue();
        });

        // 在futureWithResult操作完成后，对结果进行消费处理
        futureWithResult.thenAccept(result -> System.out.println("Result of async operation: " + result));

        // 异步执行消费者操作
        FunctionalUtils.runAsync("Hello, World!", message -> {
            System.out.println("Async consumer executed with message: " + message);
        });

        // 异步执行函数操作并返回结果
        CompletableFuture<String> futureTransformed = FunctionalUtils.applyAsync(42, number -> {
            System.out.println("Async function executed with number: " + number);
            return "Transformed number: " + (number * 2);
        });

        // 异步执行操作，并在操作完成时进行回调
        FunctionalUtils.whenCompleteAsync(() -> "Operation completed", (result, throwable) -> {
            if (throwable == null) {
                System.out.println("Async operation completed with result: " + result);
            } else {
                System.out.println("Async operation failed with exception: " + throwable);
            }
        });

        // 执行事务性操作
        boolean transactionResult = FunctionalUtils.performTransaction(
                () -> performStep1(),
                () -> performStep2(),
                () -> performStep3()
        );
        System.out.println("Transaction result: " + transactionResult);

        // 等待异步操作完成，这里只是为了示例需要
        // 在实际应用中应该以更合适的方式等待异步操作的完成
        awaitCompletionOfFutures(future, futureWithResult, futureTransformed);

        // 应用程序关闭时，关闭线程池资源
        FunctionalUtils.shutdownExecutorService();
    }

    private static void performSomeLongRunningOperation() {
        // 模拟长时间运行的操作
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private static int calculateSomeValue() {
        // 模拟计算操作
        return 42;
    }

    private static boolean performStep1() {
        // 模拟事务步骤1
        return true; // 假设步骤成功
    }

    private static boolean performStep2() {
        // 模拟事务步骤2
        return true; // 假设步骤成功
    }

    private static boolean performStep3() {
        // 模拟事务步骤3
        return true; // 假设步骤成功
    }

    @SafeVarargs
    private static void awaitCompletionOfFutures(CompletableFuture<?>... futures) {
        CompletableFuture.allOf(futures).join();
    }
}

展示了如何使用FunctionalUtils类来异步执行操作、处理返回结果、执行消费者操作、执行函数操作并返回结果、在操作完成时进行回调以及执行事务性操作。

• runAsync: 异步执行长时间运行的操作。
• supplyAsync: 异步执行有返回值的操作。
• runAsync with consumer: 异步执行消费者操作。
• applyAsync: 异步执行函数操作并返回结果。
• whenCompleteAsync: 异步执行操作并在操作完成时进行回调。
• performTransaction: 执行一系列事务性操作，如果所有步骤都成功，则返回true，否则返回false。
• awaitCompletionOfFutures: 等待所有异步操作完成，这是为了示例需要，实际应用应该有更合适的异步操作等待策略。
• shutdownExecutorService: 关闭线程池资源，应在应用程序关闭时调用。

时序图

描述：

Main代表调用FunctionalUtils方法的主线程或客户端代码。FunctionalUtils调用ExecutorService来异步执行任务。Operation代表实际执行的操作，它可能是一个Runnable、Supplier或其他函数式接口的实现。

- runAsync方法将一个Runnable提交到ExecutorService进行异步执行。
- supplyAsync方法将一个Supplier提交到ExecutorService，执行后返回一个CompletableFuture对象。
- performTransaction方法依次执行多个操作，每个操作返回一个布尔值表示成功或失败，所有操作成功则事务成功，否则事务失败。

异步处理网络请求

在处理网络请求时，异步执行可以避免I/O操作阻塞主线程。使用FunctionalUtils的supplyAsync方法，可以轻松地将网络请求任务提交给线程池，并在请求完成时处理结果。

数据流处理

对于数据流的处理，FunctionalUtils提供了map和filter方法，它们可以对集合进行转换和过滤。这些方法利用Java的流API，允许开发者以声明式的方式处理数据。

事务性文件操作

在执行一系列文件操作时，如创建、写入和删除文件，可能需要这些操作要么全部成功要么全部失败。FunctionalUtils的performTransaction方法可以保证这种一致性。

性能考虑

尽管FunctionalUtils增加了编程的便利性，但在使用异步和多线程功能时，性能和资源管理仍然是关键考虑因素。FunctionalUtils使用了一个缓存线程池来优化任务的执行，但需要在应用程序关闭时调用shutdownExecutorService方法来释放线程池资源，避免潜在的资源泄露。

结论

FunctionalUtils是一个强大的工具类，它利用Java函数式编程的特性，提供了一系列的实用功能，帮助开发者以更高效、更优雅的方式编写代码。通过结合同步、异步和事务处理功能，FunctionalUtils能够应对复杂的业务场景，同时保持代码的简洁性和可读性。