ForkJoinPool中invoke()与get()的差异及代码异常解析

问题场景 #

你提供的代码中，使用自定义ForkJoinPool执行包含parallelStream的任务时，调用invoke()会导致AtomicInteger的输出出现负数，换成get()则不会。相关代码与输出如下：

测试代码 #

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

public class TestSync {

    public static void main(String[] args) {
        List<Integer> intArray = IntStream.rangeClosed(1, 500).boxed().toList();

        System.out.println("Element size: " + intArray.size());
        
        AtomicInteger i = new AtomicInteger(intArray.size());
        ForkJoinPool customThreadPool = new ForkJoinPool(50);
        
        customThreadPool.submit( () -> {
            intArray.parallelStream().forEach(sp -> {
                System.out.println("Remaining int: " + i.decrementAndGet());
            });
        }).invoke();
        customThreadPool.shutdown();
    }

}

异常输出（invoke()时） #

Element size: 500
Remaining int: 499
Remaining int: 498
...
Remaining int: -74
Remaining int: -75
...

核心区别解析 #

1. invoke()的行为 #

invoke()是ForkJoinTask的专属方法，核心特点：

• 当前线程直接参与任务执行：调用invoke()时，当前线程（这里是main线程）会被临时纳入目标ForkJoinPool的工作线程池，和池内线程协同完成任务。
• 易导致任务重复执行：如果已经通过submit()将任务提交到池，再调用invoke()会让当前线程再次执行同一个任务——相当于同一个任务被池内线程和当前线程各执行一次。

2. get()的行为 #

get()是Future接口的标准方法，作用单纯：

• 仅阻塞等待结果：调用get()时，当前线程只会进入阻塞状态，等待池内线程完成提交的任务，不会参与任何任务的执行逻辑。

结合代码的具体分析 #

• invoke()出现负数的原因：
  你先通过submit()把包含parallelStream的Runnable提交到自定义池，池内50个线程已经开始执行这个任务（对500个元素执行decrementAndGet()）；紧接着调用invoke()，又让main线程加入池，再次执行同一个Runnable，再次启动parallelStream处理500个元素。两次执行总共让AtomicInteger被减了约1000次，初始值500最终变成负数。

• get()正常的原因：
  submit()提交任务后，只有池内线程执行一次parallelStream，完成500次decrementAndGet()；get()只是让main线程等待任务完成，不会重复执行任务，所以AtomicInteger最终减到0，不会出现负数。

正确使用方式 #

提交任务并等待完成时，二选一即可，不要同时使用submit()和invoke()：

• 方式一：提交后等待结果

  customThreadPool.submit(/* 任务逻辑 */).get();
  

• 方式二：直接执行并等待完成

  customThreadPool.invoke(/* 任务逻辑 */);
  

内容的提问来源于stack exchange，提问作者betta7391