JavaFuture并发管理避坑全攻略

Future的本质与常见误区

在Java并发编程中，java.util.concurrent.Future接口代表了一个异步计算的结果。当我们将一个任务提交给ExecutorService时，submit()方法通常会返回一个Future对象。这个Future对象充当了异步操作结果的占位符，我们可以通过它来检查任务是否完成、取消任务，以及在任务完成后获取其结果。

核心特性：

1. 结果占位符： Future本身不是数据容器，而是对某个最终计算结果的引用。
2. 不可变性： 一旦Future所代表的计算完成并返回了结果，这个结果在Future内部是不可变的。你不能通过Future对象本身去修改它所持有的结果值。
3. 类型匹配： Future的get()方法返回类型T，但Future对象本身仍然是Future类型。

常见误区：

许多初学者容易将List>误解为一个可以直接存储和修改Integer值的列表。例如，在尝试执行elements.set(firstIndex, elements.get(firstIndex).get() - randomAmount);时，会遇到编译错误。这是因为List>期望set()方法的第二个参数是一个Future类型的对象，而不是一个Integer值。即使我们能传入一个Future，也无法直接修改其内部已完成的结果。

原始代码片段中的问题如下：

List> elements = new ArrayList<>();
// ... 初始化 elements
// 尝试修改元素值，但类型不匹配
if (elements.get(firstIndex).get() - randomAmount > 0) {
    // 编译错误：set(int, Future) cannot be applied to (int, int)
    elements.set(firstIndex, elements.get(firstIndex).get() - randomAmount);
}

这里的根本问题在于，elements.get(firstIndex).get()返回的是一个Integer，而List>的set方法期望的是一个Future。此外，即使能传入，Future的不可变性也意味着无法直接“更新”一个已存在的Future所持有的值。

选择正确的共享数据结构

当目标是存储并并发修改一组整数值时，不应使用List>。我们应该根据需求选择合适的、能够直接存储和修改整数值的数据结构。

1. 直接存储整数的列表

如果只是为了存储整数，最直接的方式是使用List或int[]。

var ex = Executors.newFixedThreadPool(10);
List elements = new ArrayList<>(); // 直接存储整数
for (int i = 0; i < 100; i++) {
    elements.add(1000); // 初始化为1000
}

// 计算初始和
int sum = 0;
for (int el : elements) {
    sum += el;
}
System.out.println("Initial sum: " + sum);

// 提交任务进行并发修改
for (int i = 0; i < 10_000; i++) {
    ex.submit(() -> {
        int firstIndex = ThreadLocalRandom.current().nextInt(100);
        int secondIndex = ThreadLocalRandom.current().nextInt(100); // secondIndex在此逻辑中未使用
        int randomAmount = ThreadLocalRandom.current().nextInt(1000);

        // 注意：ArrayList在并发修改下非线程安全
        // 这里的get/set操作可能导致数据不一致或运行时错误
        if (elements.get(firstIndex) - randomAmount > 0) {
            elements.set(firstIndex, elements.get(firstIndex) - randomAmount);
        }
    });
}
// ... 后续需要关闭ExecutorService

注意事项： 尽管上述代码解决了Future的类型不匹配问题，但ArrayList本身并不是线程安全的。当多个线程同时对ArrayList进行get和set操作时，可能导致数据竞争，产生不正确的结果，甚至抛出IndexOutOfBoundsException等运行时异常。

2. 并发安全的整数数组：AtomicIntegerArray

为了在多线程环境下安全地修改共享的整数数组，推荐使用java.util.concurrent.atomic.AtomicIntegerArray。它提供了原子性的get、set、addAndGet等操作，无需显式加锁即可保证数据一致性。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerArray;

public class ConcurrentArrayUpdate {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService ex = Executors.newFixedThreadPool(10);

        // 使用AtomicIntegerArray来安全地存储和修改整数
        AtomicIntegerArray elements = new AtomicIntegerArray(100);
        for (int i = 0; i < 100; i++) {
            elements.set(i, 1000); // 初始化每个元素为1000
        }

        // 计算初始和
        long initialSum = 0;
        for (int i = 0; i < elements.length(); i++) {
            initialSum += elements.get(i);
        }
        System.out.println("Initial sum: " + initialSum);

        // 提交10,000个任务进行并发修改
        for (int i = 0; i < 10_000; i++) {
            ex.submit(() -> {
                int firstIndex = ThreadLocalRandom.current().nextInt(100);
                // int secondIndex = ThreadLocalRandom.current().nextInt(100); // 在此逻辑中未使用
                int randomAmount = ThreadLocalRandom.current().nextInt(1000);

                // 原子性地检查并更新值
                // 使用compareAndSet或循环结合get/set来确保原子性操作
                while (true) {
                    int currentValue = elements.get(firstIndex);
                    if (currentValue - randomAmount > 0) {
                        // 尝试原子性地更新值
                        if (elements.compareAndSet(firstIndex, currentValue, currentValue - randomAmount)) {
                            break; // 更新成功，退出循环
                        }
                    } else {
                        break; // 条件不满足，无需更新，退出循环
                    }
                }
            });
        }

        // 关闭ExecutorService
        ex.shutdown();
        // 等待所有任务完成，最多等待1分钟
        if (!ex.awaitTermination(1, TimeUnit.MINUTES)) {
            System.err.println("ExecutorService did not terminate in the specified time.");
        }

        // 计算最终和
        long finalSum = 0;
        for (int i = 0; i < elements.length(); i++) {
            finalSum += elements.get(i);
        }
        System.out.println("Final sum: " + finalSum);
    }
}

在上述代码中，我们使用了AtomicIntegerArray来替代List。compareAndSet方法是实现乐观锁的关键，它会检查当前值是否与期望值相等，如果相等则进行更新。这有效避免了在多线程环境下直接get然后set可能导致的数据丢失问题。

ExecutorService生命周期管理

在并发编程中，正确管理ExecutorService的生命周期至关重要。原始代码中存在过早调用ex.shutdown()的问题，这会导致在提交第一批任务后，ExecutorService立即开始关闭，后续的10,000个修改任务将无法被提交。

正确的ExecutorService使用流程：

1. 创建ExecutorService： 使用Executors工厂方法创建线程池。
2. 提交所有任务： 在所有任务都提交完毕之前，不要调用shutdown()。
3. 关闭ExecutorService： 调用shutdown()方法。这会阻止新的任务提交，并允许已提交的任务继续执行。
4. 等待任务完成： 调用awaitTermination()方法。这会阻塞当前线程，直到所有任务完成执行，或者达到超时时间，或者当前线程被中断。这是确保所有异步操作都已完成的推荐方式。
5. （可选）强制关闭： 如果awaitTermination()超时，可以使用shutdownNow()尝试中断正在执行的任务并强制关闭线程池。

在上述AtomicIntegerArray的示例代码中，我们已经演示了正确的ExecutorService关闭流程。

总结与最佳实践

1. 理解Future的用途： Future是异步计算结果的句柄，不是可变的存储容器。不要试图通过Future对象直接修改其内部结果。
2. 选择线程安全的数据结构： 当多个线程需要并发地读取和修改共享数据时，务必使用线程安全的数据结构。对于整数数组，AtomicIntegerArray是高效且推荐的选择。对于其他复杂对象或集合，可能需要使用ConcurrentHashMap、CopyOnWriteArrayList或通过synchronized块、ReentrantLock等机制进行同步。
3. 正确管理ExecutorService生命周期： 确保在所有任务提交完毕后再调用shutdown()，并使用awaitTermination()等待任务完成，以避免资源泄露或任务未执行完毕就关闭的问题。
4. 避免不必要的复杂性： 如果你的目标仅仅是存储和修改整数，而不需要异步获取单个初始值的计算结果，那么一开始就不需要使用ExecutorService和Future来初始化数据。直接创建AtomicIntegerArray并初始化即可。

通过遵循这些原则，可以有效地避免Java并发编程中常见的陷阱，编写出健壮、高效且正确的并发代码。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《JavaFuture并发管理避坑全攻略》文章吧，也可关注golang学习网公众号了解相关技术文章。