JAVA CompletableFuture异步超时控制与线程泄漏风险排查

JAVA CompletableFuture 异步超时控制与线程泄漏风险排查

各位好，今天我们来聊聊Java CompletableFuture在异步编程中超时控制和线程泄漏这两个关键问题。CompletableFuture作为Java 8引入的异步编程利器，极大地简化了并发编程，但如果不正确使用，很容易导致超时问题无法处理，甚至造成线程泄漏，最终拖垮系统。

一、CompletableFuture 的超时控制

在异步操作中，超时控制是至关重要的。我们需要在一定时间内获得结果，否则就认为操作失败。CompletableFuture 提供了多种方式来实现超时控制：

1. 使用 orTimeout() 方法 (Java 9+)

orTimeout(long timeout, TimeUnit unit) 是 Java 9 引入的方法，它允许我们指定一个超时时间。如果在指定的时间内 CompletableFuture 没有完成，它将会完成并抛出一个 TimeoutException。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class TimeoutExample {

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟一个耗时操作
                Thread.sleep(5000);
                return "Result from long operation";
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return "Interrupted";
            }
        });

        future.orTimeout(2, TimeUnit.SECONDS)
                .thenAccept(result -> System.out.println("Result: " + result))
                .exceptionally(ex -> {
                    if (ex instanceof TimeoutException) {
                        System.out.println("Timeout occurred!");
                    } else {
                        System.out.println("Exception: " + ex.getMessage());
                    }
                    return null;
                });

        // 避免主线程过早退出，导致异步任务无法执行
        try {
            Thread.sleep(3000); // 确保 orTimeout 有机会触发
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

在这个例子中，我们设置了 2 秒的超时时间。由于 Thread.sleep(5000) 会阻塞 5 秒，所以 orTimeout() 会触发，exceptionally() 会捕获到 TimeoutException。

注意：orTimeout() 仅仅是让 CompletableFuture 抛出异常，它并不会中断正在执行的任务。 如果任务还在后台运行，它仍然会消耗资源。因此，我们需要考虑如何取消或中断长时间运行的任务。

2. 使用 completeOnTimeout() 方法 (Java 9+)

completeOnTimeout(T value, long timeout, TimeUnit unit) 允许我们在超时后设置一个默认值。如果在指定的时间内 CompletableFuture 没有完成，它将会使用给定的值完成。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class CompleteOnTimeoutExample {

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟一个耗时操作
                Thread.sleep(5000);
                return "Result from long operation";
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return "Interrupted";
            }
        });

        future.completeOnTimeout("Default Value", 2, TimeUnit.SECONDS)
                .thenAccept(result -> System.out.println("Result: " + result));

        // 避免主线程过早退出，导致异步任务无法执行
        try {
            Thread.sleep(3000); // 确保 completeOnTimeout 有机会触发
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

在这个例子中，如果在 2 秒内没有获得结果，CompletableFuture 将会使用 "Default Value" 完成。

同样，completeOnTimeout() 也不会中断正在执行的任务。

3. 手动实现超时控制

如果我们需要更精细的控制，例如在超时后取消任务，我们可以手动实现超时控制。 这通常涉及创建一个单独的线程来监控 CompletableFuture 的完成状态，并在超时后执行取消操作。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ManualTimeoutExample {

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟一个耗时操作
                Thread.sleep(5000);
                return "Result from long operation";
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return "Interrupted";
            }
        });

        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

        scheduler.schedule(() -> {
            if (!future.isDone()) {
                System.out.println("Timeout occurred. Cancelling the task.");
                future.cancel(true); // 尝试中断任务
            }
        }, 2, TimeUnit.SECONDS);

        future.thenAccept(result -> System.out.println("Result: " + result))
                .exceptionally(ex -> {
                    System.out.println("Exception: " + ex.getMessage());
                    return null;
                });

        // 避免主线程过早退出，导致异步任务无法执行
        try {
            Thread.sleep(3000); // 确保 scheduler 有机会触发
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        scheduler.shutdown();
    }
}

在这个例子中，我们使用 ScheduledExecutorService 来安排一个任务，该任务在 2 秒后检查 CompletableFuture 是否完成。 如果没有完成，我们就调用 future.cancel(true) 来尝试中断任务。

注意： future.cancel(true) 只是尝试中断任务。 如果任务正在执行一些无法中断的操作（例如，正在进行 I/O 操作），它可能无法被中断。

超时控制策略选择

超时控制方法	优点	缺点	适用场景
orTimeout()	简单易用，代码简洁	不会中断正在执行的任务	只需要抛出超时异常，不需要中断任务。
completeOnTimeout()	简单易用，可以设置默认值	不会中断正在执行的任务	只需要使用默认值完成 CompletableFuture，不需要中断任务。
手动实现	灵活性高，可以实现更复杂的超时处理逻辑，例如取消任务	代码复杂，需要手动管理线程	需要中断任务，或者需要执行其他复杂的超时处理逻辑。

二、CompletableFuture 的线程泄漏风险

CompletableFuture 依赖于线程池来执行异步任务。 如果线程池配置不当，或者使用方式不正确，很容易导致线程泄漏。

1. 默认线程池的问题

如果我们在创建 CompletableFuture 时没有指定 Executor，它将会使用 ForkJoinPool.commonPool()。 这是一个全局的共享线程池，所有 CompletableFuture 都会使用它。

ForkJoinPool.commonPool() 的一个问题是，它的线程数量是有限的，而且它的线程是守护线程。 如果我们的任务阻塞或者长时间运行，可能会耗尽线程池中的所有线程，导致其他任务无法执行。 此外，由于守护线程的特性，如果主线程退出，守护线程也会被终止，导致任务无法完成。

2. 线程泄漏的常见原因

• 任务阻塞： 如果任务因为 I/O 操作或者其他原因阻塞，线程将会被阻塞，无法执行其他任务。
• 长时间运行的任务： 如果任务需要很长时间才能完成，线程将会被占用，无法执行其他任务。
• 未处理的异常： 如果任务抛出异常，但是没有被捕获，可能会导致线程池中的线程被终止。
• 错误的线程池配置： 如果线程池的线程数量太少，或者队列容量太小，可能会导致线程池无法处理所有的任务。
• 循环依赖： 多个 CompletableFuture 相互依赖，形成循环依赖，导致任务无法完成，线程一直被占用。

3. 如何避免线程泄漏

• 使用自定义线程池： 避免使用 ForkJoinPool.commonPool()，而是使用自定义的 ExecutorService。 这样可以更好地控制线程池的配置，例如线程数量、队列容量、拒绝策略等。

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

public class CustomExecutorExample {

    public static void main(String[] args) {
        // 创建一个固定大小的线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟一个耗时操作
                Thread.sleep(2000);
                return "Result from long operation";
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return "Interrupted";
            }
        }, executor);

        future.thenAccept(result -> System.out.println("Result: " + result));

        // 关闭线程池
        executor.shutdown();

        // 避免主线程过早退出，导致异步任务无法执行
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

• 避免阻塞操作： 尽量避免在 CompletableFuture 的任务中使用阻塞操作。 如果必须使用阻塞操作，可以考虑使用异步 I/O 或者将阻塞操作放在单独的线程中执行。
• 处理异常： 务必捕获 CompletableFuture 中的异常，避免线程被终止。 可以使用 exceptionally() 方法或者 handle() 方法来处理异常。
• 设置合理的超时时间： 设置合理的超时时间，避免任务长时间运行占用线程。
• 避免循环依赖： 仔细检查 CompletableFuture 之间的依赖关系，避免出现循环依赖。
• 监控线程池： 使用 JMX 或者其他监控工具来监控线程池的状态，例如线程数量、活跃线程数量、队列长度等。 如果发现线程池出现异常，及时进行处理。
• 使用有界队列： 在自定义线程池时，使用有界队列，例如 ArrayBlockingQueue 或 LinkedBlockingQueue(capacity)。 当队列满时，拒绝提交新的任务，可以防止 OOM 错误。 可以自定义 RejectedExecutionHandler 来处理被拒绝的任务，例如记录日志或执行重试。

import java.util.concurrent.*;

public class BoundedQueueExample {

    public static void main(String[] args) {
        // 创建一个固定大小的线程池，使用有界队列
        int corePoolSize = 5;
        int maxPoolSize = 10;
        long keepAliveTime = 60L;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(100); // 使用 ArrayBlockingQueue，设置容量为 100
        RejectedExecutionHandler rejectedExecutionHandler = new CustomRejectedExecutionHandler(); // 自定义拒绝策略

        ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, rejectedExecutionHandler);

        // 提交任务
        for (int i = 0; i < 200; i++) {
            int taskNumber = i;
            executor.execute(() -> {
                try {
                    System.out.println("Executing task: " + taskNumber + " in thread: " + Thread.currentThread().getName());
                    Thread.sleep(100); // 模拟耗时操作
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        executor.shutdown();
    }

    // 自定义拒绝策略
    static class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            System.err.println("Task rejected: " + r.toString() + " Executor is shutdown: " + executor.isShutdown());
            // 在这里可以执行重试、记录日志等操作
            // 例如：将任务重新放入队列
            // try {
            //     executor.getQueue().put(r);
            // } catch (InterruptedException e) {
            //     Thread.currentThread().interrupt();
            //     System.err.println("Failed to re-queue task: " + r.toString());
            // }
        }
    }
}

• 使用 try-finally 确保资源释放: 在 CompletableFuture 的任务中，确保在使用完资源后，及时释放资源。 例如，关闭数据库连接、文件流等。 可以使用 try-finally 语句来确保资源总是被释放，即使任务抛出异常。

import java.util.concurrent.CompletableFuture;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

public class ResourceCleanupExample {

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            Connection connection = null;
            try {
                // 模拟获取数据库连接
                connection = DriverManager.getConnection("jdbc:your_database_url", "username", "password");
                // 执行数据库操作
                return "Database operation successful";
            } catch (SQLException e) {
                System.err.println("Database error: " + e.getMessage());
                return "Database operation failed";
            } finally {
                // 确保连接被关闭
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (SQLException e) {
                        System.err.println("Error closing connection: " + e.getMessage());
                    }
                }
            }
        });

        future.thenAccept(result -> System.out.println("Result: " + result));

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

4. 线程泄漏风险排查

如果怀疑存在线程泄漏，可以使用以下方法进行排查：

• 使用 JConsole 或者 VisualVM： 这些工具可以监控 JVM 的线程状态，可以查看线程数量、线程堆栈等信息。 如果发现线程数量不断增加，或者存在大量处于 BLOCKED 或者 WAITING 状态的线程，可能存在线程泄漏。

• 使用线程转储（Thread Dump）： 线程转储可以生成 JVM 中所有线程的快照。 通过分析线程转储，可以找到阻塞或者长时间运行的线程，以及它们的调用堆栈。 可以使用 jstack 命令或者 JConsole/VisualVM 来生成线程转储。

  jstack <pid> > thread_dump.txt

  然后分析 thread_dump.txt 文件，查找可能存在问题的线程。 关注以下信息：

  • 线程的状态（NEW, RUNNABLE, BLOCKED, WAITING, TIMED_WAITING, TERMINATED）
  • 线程的调用堆栈
  • 线程持有的锁

• 代码审查： 仔细检查代码，特别是涉及到 CompletableFuture 的部分，查找可能导致线程泄漏的原因。 重点关注线程池的使用、阻塞操作、异常处理、资源释放等方面。

• 使用性能分析工具： 使用性能分析工具，例如 JProfiler 或者 YourKit，可以更详细地分析线程的执行情况，找到性能瓶颈和潜在的线程泄漏问题。

三、实际案例分析

假设我们有一个订单处理系统，需要调用多个外部服务来完成订单处理。 如果其中一个外部服务响应缓慢或者出现故障，可能会导致订单处理超时，甚至导致线程泄漏。

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

public class OrderProcessingExample {

    private static final ExecutorService executor = Executors.newFixedThreadPool(10);

    public static void main(String[] args) {
        String orderId = "12345";

        CompletableFuture<String> future = processOrder(orderId);

        future.thenAccept(result -> System.out.println("Order processing result: " + result))
                .exceptionally(ex -> {
                    System.err.println("Order processing failed: " + ex.getMessage());
                    return null;
                });

        executor.shutdown();
        try {
            executor.awaitTermination(10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static CompletableFuture<String> processOrder(String orderId) {
        CompletableFuture<String> validateOrderFuture = CompletableFuture.supplyAsync(() -> validateOrder(orderId), executor);
        CompletableFuture<String> checkInventoryFuture = validateOrderFuture.thenCompose(result -> CompletableFuture.supplyAsync(() -> checkInventory(orderId), executor));
        CompletableFuture<String> processPaymentFuture = checkInventoryFuture.thenCompose(result -> CompletableFuture.supplyAsync(() -> processPayment(orderId), executor));
        CompletableFuture<String> shipOrderFuture = processPaymentFuture.thenCompose(result -> CompletableFuture.supplyAsync(() -> shipOrder(orderId), executor));

        return shipOrderFuture.orTimeout(5, TimeUnit.SECONDS); // 设置超时时间
    }

    private static String validateOrder(String orderId) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Validating order: " + orderId);
        return "Order validated";
    }

    private static String checkInventory(String orderId) {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Checking inventory for order: " + orderId);
        return "Inventory checked";
    }

    private static String processPayment(String orderId) {
        try {
            Thread.sleep(6000); // 模拟支付服务响应缓慢
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Processing payment for order: " + orderId);
        return "Payment processed";
    }

    private static String shipOrder(String orderId) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Shipping order: " + orderId);
        return "Order shipped";
    }
}

在这个例子中，processPayment() 方法模拟支付服务响应缓慢，导致整个订单处理超时。 我们使用了 orTimeout() 方法来设置超时时间，如果订单处理超过 5 秒，将会抛出 TimeoutException。 同时使用了自定义的线程池来执行异步任务，并设置了线程池的关闭策略。

优化建议：

• 熔断机制： 对于不稳定的外部服务，可以引入熔断机制，防止服务雪崩。
• 降级策略： 在超时或者服务不可用时，可以采用降级策略，例如返回缓存数据或者执行默认操作。
• 异步重试： 对于可以重试的操作，可以采用异步重试机制，提高系统的可用性。

四、总结

CompletableFuture 提供了强大的异步编程能力，但也需要谨慎使用，避免超时问题和线程泄漏。 通过合理设置超时时间、使用自定义线程池、避免阻塞操作、处理异常、监控线程池等方法，可以有效地避免这些问题。 在实际开发中，要根据具体的业务场景选择合适的超时控制策略和线程池配置，并持续监控系统的运行状态，及时发现和解决问题。

五、总结一下

• 合理使用 orTimeout 和 completeOnTimeout 可以控制异步任务的超时，但不会中断任务。
• 避免使用默认线程池，使用自定义线程池可以更好地控制资源。
• 排查线程泄漏需要关注线程状态、调用堆栈和资源释放。