JAVA CompletableFuture深度使用中的资源竞争与同步风险排查

好的，我们开始。

JAVA CompletableFuture 深度使用中的资源竞争与同步风险排查

大家好，今天我们深入探讨 Java CompletableFuture 的高级用法，特别是它在并发编程中可能遇到的资源竞争和同步风险，以及如何有效地排查和解决这些问题。CompletableFuture 提供了强大的异步编程模型，但如果不小心使用，很容易引入难以调试的并发 bug。

CompletableFuture 简介与基础

首先，简单回顾一下 CompletableFuture 的核心概念。CompletableFuture 代表一个异步计算的结果，它允许你以非阻塞的方式组合、链式调用多个异步操作。

核心概念：

异步执行： CompletableFuture 可以在不同的线程中执行任务，避免阻塞主线程。链式调用： 可以使用 thenApply, thenAccept, thenCompose 等方法将多个 CompletableFuture 连接起来，形成一个异步处理流水线。异常处理： 提供了 exceptionally, handle 等方法来处理异步操作中可能出现的异常。组合操作： 可以使用 allOf, anyOf 等方法将多个 CompletableFuture 组合成一个，等待所有任务完成或任意一个任务完成。

示例代码：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
 
public class CompletableFutureExample {
 
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Hello, CompletableFuture!";
        });
 
        // 对结果进行转换
        CompletableFuture<String> transformedFuture = future.thenApply(result -> result.toUpperCase());
 
        // 获取最终结果（会阻塞，仅用于演示）
        String result = transformedFuture.get();
        System.out.println(result); // 输出: HELLO, COMPLETABLEFUTURE!
    }
}

这个例子展示了 CompletableFuture 的基本用法：使用 supplyAsync 创建一个异步任务，然后使用 thenApply 对结果进行转换。

资源竞争的场景与风险

在使用 CompletableFuture 进行并发编程时，最常见的风险之一就是资源竞争。当多个 CompletableFuture 试图同时访问和修改同一个共享资源时，如果没有适当的同步机制，就会出现数据不一致、死锁等问题。

常见场景：

共享变量的并发修改： 多个 CompletableFuture 同时修改一个静态变量或实例变量。数据库连接池的竞争： 多个 CompletableFuture 同时从数据库连接池获取连接。缓存的并发更新： 多个 CompletableFuture 同时更新缓存中的数据。文件 I/O 的并发访问： 多个 CompletableFuture 同时读写同一个文件。

示例代码（资源竞争）：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
 
public class ResourceContentionExample {
 
    private static int counter = 0;
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
 
        CompletableFuture<?>[] futures = IntStream.range(0, 1000)
                .mapToObj(i -> CompletableFuture.runAsync(() -> {
                    // 模拟耗时操作
                    for (int j = 0; j < 1000; j++) {
                        counter++; //  increment counter
                    }
                }, executor))
                .toArray(CompletableFuture<?>[]::new);
 
        CompletableFuture.allOf(futures).join();
 
        executor.shutdown();
        System.out.println("Counter value: " + counter); // 期望输出 1000000，但结果通常小于该值
    }
}

在这个例子中，多个 CompletableFuture 并发地增加 counter 变量的值。由于没有同步机制，counter++ (实际是读-改-写操作) 不是原子操作，导致多个线程可能读取到相同的值，然后同时对其进行修改，从而丢失更新。最终的结果通常小于期望值 1000000。

同步机制与解决方案

为了避免资源竞争，我们需要使用适当的同步机制来保护共享资源。

常见的同步机制：

synchronized 关键字： Java 内置的同步机制，可以保证同一时刻只有一个线程可以访问被 synchronized 修饰的代码块或方法。ReentrantLock： 可重入锁，提供了比 synchronized 关键字更灵活的锁机制，例如可以实现公平锁、可中断锁等。AtomicInteger 等原子类： 提供了原子操作，可以保证对单个变量的读-改-写操作是原子性的。ConcurrentHashMap 等并发容器： 提供了线程安全的集合类，可以在并发环境下安全地进行读写操作。Semaphore： 信号量，可以控制同时访问某个资源的线程数量。

示例代码（使用 AtomicInteger 解决资源竞争）：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
 
public class AtomicIntegerExample {
 
    private static AtomicInteger counter = new AtomicInteger(0);
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
 
        CompletableFuture<?>[] futures = IntStream.range(0, 1000)
                .mapToObj(i -> CompletableFuture.runAsync(() -> {
                    // 模拟耗时操作
                    for (int j = 0; j < 1000; j++) {
                        counter.incrementAndGet(); // 使用原子操作
                    }
                }, executor))
                .toArray(CompletableFuture<?>[]::new);
 
        CompletableFuture.allOf(futures).join();
 
        executor.shutdown();
        System.out.println("Counter value: " + counter.get()); // 输出 1000000
    }
}

在这个例子中，我们使用 AtomicInteger 来代替普通的 int 变量。AtomicInteger.incrementAndGet() 方法提供了原子性的自增操作，保证了多个线程可以安全地修改 counter 变量的值。

示例代码（使用 synchronized 解决资源竞争）：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
 
public class SynchronizedExample {
 
    private static int counter = 0;
    private static final Object lock = new Object(); // 用于同步的锁对象
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
 
        CompletableFuture<?>[] futures = IntStream.range(0, 1000)
                .mapToObj(i -> CompletableFuture.runAsync(() -> {
                    // 模拟耗时操作
                    for (int j = 0; j < 1000; j++) {
                        synchronized (lock) { // 使用 synchronized 关键字
                            counter++;
                        }
                    }
                }, executor))
                .toArray(CompletableFuture<?>[]::new);
 
        CompletableFuture.allOf(futures).join();
 
        executor.shutdown();
        System.out.println("Counter value: " + counter); // 输出 1000000
    }
}

在这个例子中，我们使用 synchronized 关键字来保护对 counter 变量的访问。只有获取到 lock 对象的锁的线程才能访问 synchronized 代码块，从而避免了资源竞争。

选择同步机制的原则：

原子性： 如果只需要对单个变量进行原子操作，优先选择 AtomicInteger 等原子类。性能： AtomicInteger 等原子类的性能通常比 synchronized 关键字更好，因为它们使用了更底层的 CAS (Compare and Swap) 指令。灵活性： 如果需要更灵活的锁机制，例如公平锁、可中断锁等，可以选择 ReentrantLock。复杂性： synchronized 关键字使用起来更简单，但功能相对有限。读多写少： 对于读多写少的场景，可以考虑使用 ReadWriteLock，它允许多个线程同时读取共享资源，但只允许一个线程写入共享资源。

死锁的预防与排查

死锁是指两个或多个线程互相等待对方释放资源，导致所有线程都无法继续执行的情况。在使用 CompletableFuture 进行并发编程时，死锁也是一个需要特别注意的问题。

死锁的四个必要条件：

互斥条件： 资源只能被一个线程占用。占有且等待条件： 线程占有资源，并等待其他线程释放资源。不可剥夺条件： 线程已经占有的资源不能被强制剥夺。循环等待条件： 多个线程形成循环等待资源的关系。

示例代码（死锁）：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class DeadlockExample {
 
    private static final Object lock1 = new Object();
    private static final Object lock2 = new Object();
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
 
        CompletableFuture.runAsync(() -> {
            synchronized (lock1) {
                System.out.println("Thread 1: Holding lock1...");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Thread 1: Waiting for lock2...");
                synchronized (lock2) {
                    System.out.println("Thread 1: Holding lock1 & lock2...");
                }
            }
        }, executor);
 
        CompletableFuture.runAsync(() -> {
            synchronized (lock2) {
                System.out.println("Thread 2: Holding lock2...");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Thread 2: Waiting for lock1...");
                synchronized (lock1) {
                    System.out.println("Thread 2: Holding lock2 & lock1...");
                }
            }
        }, executor);
 
        executor.shutdown();
        Thread.sleep(1000); // Allow time for deadlock to occur
    }
}

在这个例子中，线程 1 先获取 lock1，然后尝试获取 lock2；线程 2 先获取 lock2，然后尝试获取 lock1。由于两个线程互相等待对方释放锁，导致死锁。程序会一直阻塞，无法继续执行。

预防死锁的策略：

避免循环等待： 打破循环等待条件，例如规定所有线程必须按照固定的顺序获取锁。限制资源占用时间： 尽量缩短线程占用资源的时间，减少死锁发生的概率。使用超时机制： 在获取锁时设置超时时间，如果超过超时时间仍然无法获取锁，则放弃获取，释放已经占有的资源。使用死锁检测工具： 一些工具可以检测 Java 程序中的死锁，例如 VisualVM。

示例代码（使用超时机制避免死锁）：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
 
public class DeadlockAvoidanceExample {
 
    private static final ReentrantLock lock1 = new ReentrantLock();
    private static final ReentrantLock lock2 = new ReentrantLock();
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
 
        CompletableFuture.runAsync(() -> {
            try {
                if (lock1.tryLock(100, TimeUnit.MILLISECONDS)) {
                    try {
                        System.out.println("Thread 1: Holding lock1...");
                        Thread.sleep(10);
                        System.out.println("Thread 1: Waiting for lock2...");
                        if (lock2.tryLock(100, TimeUnit.MILLISECONDS)) {
                            try {
                                System.out.println("Thread 1: Holding lock1 & lock2...");
                            } finally {
                                lock2.unlock();
                            }
                        } else {
                            System.out.println("Thread 1: Failed to acquire lock2, releasing lock1...");
                        }
                    } finally {
                        lock1.unlock();
                    }
                } else {
                    System.out.println("Thread 1: Failed to acquire lock1...");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, executor);
 
        CompletableFuture.runAsync(() -> {
            try {
                if (lock2.tryLock(100, TimeUnit.MILLISECONDS)) {
                    try {
                        System.out.println("Thread 2: Holding lock2...");
                        Thread.sleep(10);
                        System.out.println("Thread 2: Waiting for lock1...");
                        if (lock1.tryLock(100, TimeUnit.MILLISECONDS)) {
                            try {
                                System.out.println("Thread 2: Holding lock2 & lock1...");
                            } finally {
                                lock1.unlock();
                            }
                        } else {
                            System.out.println("Thread 2: Failed to acquire lock1, releasing lock2...");
                        }
                    } finally {
                        lock2.unlock();
                    }
                } else {
                    System.out.println("Thread 2: Failed to acquire lock2...");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, executor);
 
        executor.shutdown();
        Thread.sleep(1000);
    }
}

在这个例子中，我们使用 ReentrantLock.tryLock() 方法来尝试获取锁，并设置了超时时间为 100 毫秒。如果超过超时时间仍然无法获取锁，则放弃获取，释放已经占有的资源，从而避免死锁。

其他同步风险与注意事项

除了资源竞争和死锁之外，还有一些其他的同步风险需要注意。

可见性问题： 在一个线程中修改了共享变量的值，其他线程可能无法立即看到最新的值。可以使用 volatile 关键字来保证变量的可见性。指令重排序： 编译器和处理器可能会对指令进行重排序，导致程序的执行顺序与代码的编写顺序不一致。可以使用 volatile 关键字或 synchronized 关键字来禁止指令重排序。线程安全问题： 确保所有共享资源都是线程安全的，或者使用适当的同步机制来保护它们。

表格总结：

风险	描述	解决方案
资源竞争	多个线程同时访问和修改同一个共享资源，导致数据不一致。	使用 `synchronized` 关键字、`ReentrantLock`、`AtomicInteger` 等同步机制来保护共享资源。
死锁	两个或多个线程互相等待对方释放资源，导致所有线程都无法继续执行。	避免循环等待、限制资源占用时间、使用超时机制、使用死锁检测工具。
可见性问题	在一个线程中修改了共享变量的值，其他线程可能无法立即看到最新的值。	使用 `volatile` 关键字来保证变量的可见性。
指令重排序	编译器和处理器可能会对指令进行重排序，导致程序的执行顺序与代码的编写顺序不一致。	使用 `volatile` 关键字或 `synchronized` 关键字来禁止指令重排序。
线程安全问题	共享资源不是线程安全的，或者没有使用适当的同步机制来保护它们。	确保所有共享资源都是线程安全的，或者使用适当的同步机制来保护它们。

CompletableFuture 线程池的选择

CompletableFuture 默认使用 ForkJoinPool.commonPool() 作为其异步任务的执行线程池。虽然方便，但在高并发场景下，共享的 commonPool 可能会成为性能瓶颈。因此，建议根据实际情况选择合适的线程池。

选择线程池的原则：

CPU 密集型任务： 使用与 CPU 核心数相等的线程池大小。I/O 密集型任务： 使用更大的线程池大小，例如 CPU 核心数的两倍或更多。任务的性质： 根据任务的性质选择合适的线程池类型，例如 FixedThreadPool、CachedThreadPool、ScheduledThreadPool 等。

示例代码（使用自定义线程池）：



import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class CustomThreadPoolExample {
 
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10);
 
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Hello, CompletableFuture!";
        }, executor); // 使用自定义线程池
 
        future.thenAccept(result -> System.out.println("Result: " + result));
 
        executor.shutdown();
    }
}

在这个例子中，我们使用 Executors.newFixedThreadPool(10) 创建了一个固定大小为 10 的线程池，并将其作为 supplyAsync 方法的参数传入，从而让 CompletableFuture 在自定义线程池中执行任务。

诊断工具与技巧

在排查 CompletableFuture 的资源竞争和同步风险时，可以使用一些诊断工具和技巧。

线程转储（Thread Dump）： 可以查看 Java 进程中所有线程的状态，包括线程是否被阻塞、正在等待哪个锁等。可以使用 jstack 命令或 VisualVM 等工具生成线程转储。性能分析工具： 可以使用 JProfiler、YourKit 等性能分析工具来分析 CompletableFuture 的性能瓶颈，例如哪些线程占用了过多的 CPU 时间、哪些锁竞争激烈等。日志记录： 在关键代码段添加日志记录，可以帮助你了解程序的执行流程，以及哪些线程访问了共享资源。单元测试： 编写并发单元测试，可以帮助你发现 CompletableFuture 的资源竞争和同步风险。可以使用 JUnit、Mockito 等测试框架。代码审查： 进行代码审查，可以帮助你发现潜在的并发 bug。