ForkJoinPool的工作窃取（Work Stealing）：平衡线程池负载的算法细节 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

ForkJoinPool 的工作窃取：平衡线程池负载的算法细节

大家好，今天我们来深入探讨 ForkJoinPool 中至关重要的工作窃取（Work Stealing）算法。ForkJoinPool 是 Java 并发包 (java.util.concurrent) 中用于执行分治任务的线程池，其高效性很大程度上依赖于工作窃取机制，它能够在多线程环境下有效地平衡任务负载，最大限度地利用 CPU 资源。

1. ForkJoinPool 的基本架构

在深入工作窃取之前，我们先简单回顾一下 ForkJoinPool 的基本架构。

ForkJoinPool: 整个线程池，负责管理 Worker 线程。
ForkJoinWorkerThread: 实际执行任务的线程。每个线程都有自己的双端队列 (Deque)。
ForkJoinTask: 代表一个可以被 ForkJoinPool 执行的任务。
Deque (双端队列): 每个 Worker 线程维护一个双端队列，用于存储待执行的 ForkJoinTask。
工作窃取队列（Work-Stealing Queue）: 实际上就是上面说的双端队列，每个线程都有自己的一个。

2. 分治模型与 ForkJoinTask

ForkJoinPool 适用于解决可以被分解为更小、独立的子任务的问题，这种模式称为分治（Divide and Conquer）。ForkJoinTask 是所有在 ForkJoinPool 中执行的任务的基类。通常，我们会继承 RecursiveTask (用于有返回值的任务) 或 RecursiveAction (用于没有返回值的任务) 来实现自己的任务。

举个例子，计算一个数组的和：

import java.util.concurrent.RecursiveTask;

class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 1000; // 阈值，决定何时进行分割
    private final long[] array;
    private final int start;
    private final int end;

    public SumTask(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= THRESHOLD) {
            // 小于阈值，直接计算
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        } else {
            // 大于阈值，分割成两个子任务
            int middle = (start + end) / 2;
            SumTask leftTask = new SumTask(array, start, middle);
            SumTask rightTask = new SumTask(array, middle, end);

            // 执行子任务
            leftTask.fork(); // 异步执行左侧任务
            rightTask.fork(); // 异步执行右侧任务

            // 合并结果
            Long leftResult = leftTask.join(); // 等待左侧任务完成
            Long rightResult = rightTask.join(); // 等待右侧任务完成
            return leftResult + rightResult;
        }
    }
}

在这个例子中，SumTask 将数组分割成更小的部分，直到小于阈值 THRESHOLD 才进行实际的求和操作。fork() 方法将子任务放入当前线程的 Deque 中，并通知 ForkJoinPool 执行该任务。join() 方法用于等待子任务完成并获取其结果。

3. 工作窃取的原理

工作窃取的核心思想是：当一个线程完成自己队列中的所有任务后，它不会空闲，而是尝试从其他线程的队列中“窃取”任务来执行。这样可以避免某些线程繁忙，而另一些线程空闲的情况，从而提高整体的并行效率。

具体步骤如下：

任务提交: 将初始任务提交到 ForkJoinPool 中。通常，初始任务会被放入某个 Worker 线程的 Deque 中。
任务分解 (Fork): 当一个 Worker 线程执行任务时，如果任务可以被分解成更小的子任务，它会使用 fork() 方法将子任务放入自己的 Deque 中。
任务执行: Worker 线程从自己的 Deque 的头部获取任务并执行。这种 LIFO (Last-In-First-Out) 的方式有助于提高缓存命中率，因为最近放入的任务很可能还在 CPU 缓存中。
任务窃取 (Steal): 当一个 Worker 线程的 Deque 为空时，它会随机选择一个其他 Worker 线程，并尝试从该线程的 Deque 的尾部窃取任务。这种 FIFO (First-In-First-Out) 的方式有助于避免 "饿死" 某些任务，因为队列尾部的任务通常是较早放入的，优先级应该较高。
任务合并 (Join): 当一个任务需要等待其子任务完成时，它会调用 join() 方法。 join() 方法会阻塞当前线程，直到子任务完成并返回结果。

4. 工作窃取的优势

负载均衡: 工作窃取能够动态地平衡线程池中的任务负载，避免某些线程过载，而另一些线程空闲的情况。
提高 CPU 利用率: 通过让空闲线程窃取任务，可以最大限度地利用 CPU 资源，提高整体的并行效率。
减少线程间的竞争: 每个线程都有自己的 Deque，减少了线程间对共享数据结构的竞争，提高了并发性能。
适应性强: 工作窃取算法能够适应不同的任务粒度和计算环境，具有较强的适应性。

5. 工作窃取算法的细节

目标选择: 当一个线程需要窃取任务时，它如何选择目标线程？ ForkJoinPool 使用伪随机数生成器来选择目标线程。这种方式简单高效，但也可能导致某些线程被频繁窃取，而另一些线程很少被窃取。
窃取操作: 窃取操作必须是原子性的，以避免多个线程同时窃取同一个任务。 ForkJoinPool 使用 CAS (Compare and Swap) 操作来保证窃取操作的原子性。
空闲线程的处理: 当一个线程找不到可以窃取的任务时，它会进入休眠状态，并定期检查是否有新的任务可以窃取。

6. 工作窃取的 Java 代码实现 (简化版)

为了更清晰地理解工作窃取的原理，我们来看一个简化的 Java 代码实现（仅用于演示，不具备完整的 ForkJoinPool 功能）：

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

class Worker implements Runnable {
    private final Deque<Runnable> deque = new ArrayDeque<>();
    private final Worker[] pool;
    private final Random random = new Random();
    private final AtomicBoolean isRunning = new AtomicBoolean(true); //标识线程是否运行

    public Worker(Worker[] pool) {
        this.pool = pool;
    }

    public void submit(Runnable task) {
        synchronized (deque) {
            deque.offerFirst(task); // 从头部加入任务
            deque.notify();
        }
    }

    @Override
    public void run() {
        while (isRunning.get()) {
            Runnable task = null;
            synchronized (deque) {
                while (deque.isEmpty()) {
                    try {
                        deque.wait(); // 等待任务
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                }
                task = deque.pollFirst(); // 从头部获取任务
            }

            if (task != null) {
                try {
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else {
                // 尝试窃取任务
                stealWork();
            }
        }
    }

    private void stealWork() {
        int targetIndex = random.nextInt(pool.length);
        Worker target = pool[targetIndex];
        if (target != this) {
            Runnable stolenTask = null;
            synchronized (target.deque) {
                if (!target.deque.isEmpty()) {
                    stolenTask = target.deque.pollLast(); // 从尾部窃取任务
                }
            }

            if (stolenTask != null) {
                try {
                    stolenTask.run();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else {
                // 没有窃取到任务，稍等片刻
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }

    public void shutdown() {
        isRunning.set(false);
        synchronized (deque) {
            deque.notifyAll();
        }
    }
}

class SimpleThreadPool {
    private final Worker[] workers;
    private final Thread[] threads;

    public SimpleThreadPool(int poolSize) {
        workers = new Worker[poolSize];
        threads = new Thread[poolSize];
        for (int i = 0; i < poolSize; i++) {
            workers[i] = new Worker(workers);
            threads[i] = new Thread(workers[i]);
            threads[i].start();
        }
    }

    public void submit(Runnable task) {
        int workerIndex = new Random().nextInt(workers.length); // 随机选择一个 Worker
        workers[workerIndex].submit(task);
    }

    public void shutdown() {
        for (Worker worker : workers) {
            worker.shutdown();
        }
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

public class SimplifiedWorkStealing {
    public static void main(String[] args) throws InterruptedException {
        int poolSize = 4;
        SimpleThreadPool pool = new SimpleThreadPool(poolSize);

        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            pool.submit(() -> {
                System.out.println("Task " + taskNumber + " executed by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        Thread.sleep(1000);
        pool.shutdown();
    }
}

这个简化的例子展示了工作窃取的基本流程： Worker 线程从自己的 Deque 中获取任务执行，如果 Deque 为空，则尝试从其他 Worker 线程的 Deque 尾部窃取任务。

7. ForkJoinPool 的配置

ForkJoinPool 的性能受到多种因素的影响，包括线程池大小、任务粒度、以及硬件配置等。合理配置 ForkJoinPool 可以最大限度地提高其性能。

线程池大小: 线程池大小应该根据 CPU 核心数和任务的计算密集程度来确定。对于计算密集型任务，线程池大小通常设置为 CPU 核心数或 CPU 核心数 + 1。对于 I/O 密集型任务，线程池大小可以适当增加。可以使用 ForkJoinPool(int parallelism) 构造函数来指定线程池大小。如果使用默认构造函数，则线程池大小等于 CPU 核心数。
任务粒度: 任务粒度是指子任务的大小。如果任务粒度过小，会导致过多的任务创建和调度开销。如果任务粒度过大，会导致负载不均衡。应该根据具体情况选择合适的任务粒度。
公共池 (Common Pool): ForkJoinPool 提供了一个公共池，可以通过 ForkJoinPool.commonPool() 方法获取。公共池适用于执行一些小的、独立的任务。但是，应该避免在公共池中执行长时间运行的任务，因为这可能会影响其他使用公共池的任务的性能。

8. 工作窃取的开销

虽然工作窃取能够提高并行效率，但也存在一定的开销：

线程间通信: 窃取任务需要进行线程间通信，这会带来一定的开销。
竞争: 多个线程可能同时尝试窃取同一个任务，这会导致竞争。
伪共享 (False Sharing): 如果多个线程访问同一个缓存行中的不同变量，会导致伪共享，从而降低性能。

9. 实际应用中的注意事项

避免阻塞操作: 在 ForkJoinTask 中应该避免执行阻塞操作，因为这会导致线程空闲，从而降低并行效率。如果必须执行阻塞操作，应该使用 ManagedBlocker 接口。
异常处理: ForkJoinTask 中抛出的异常会被封装成 ExecutionException 异常。应该在 join() 方法中捕获该异常并进行处理。
监控和调优: 可以使用 Java 的监控工具 (例如 JConsole 或 VisualVM) 来监控 ForkJoinPool 的性能，并根据监控结果进行调优。

10. 表格总结 ForkJoinPool 的关键概念

概念	描述	作用
ForkJoinPool	线程池，用于执行分治任务。	管理 Worker 线程，执行 ForkJoinTask。
ForkJoinTask	代表一个可以被 ForkJoinPool 执行的任务。	定义任务的计算逻辑，支持任务分解和合并。
Deque	双端队列，每个 Worker 线程维护一个 Deque，用于存储待执行的 ForkJoinTask。	存储任务，支持 LIFO (从头部获取) 和 FIFO (从尾部窃取) 两种访问方式。
工作窃取 (Work Stealing)	当一个 Worker 线程完成自己队列中的所有任务后，它会尝试从其他线程的队列中“窃取”任务来执行。	平衡线程池中的任务负载，提高 CPU 利用率。
Fork()	将任务放入当前线程的 Deque 中，并通知 ForkJoinPool 执行该任务。	异步执行子任务。
Join()	等待子任务完成并获取其结果。	同步等待子任务完成，合并子任务的结果。

11. 深入理解工作窃取的价值

工作窃取是 ForkJoinPool 的核心竞争力，它能够有效地利用多核 CPU 资源，提高并行效率。通过动态地平衡任务负载，避免线程空闲，从而最大限度地发挥硬件性能。了解工作窃取的原理和实现细节，能够帮助我们更好地使用 ForkJoinPool，编写高效的并行程序。

12. 合理利用，提升并发程序性能

ForkJoinPool 和工作窃取机制为我们提供了一个强大的工具，用于解决可以分解为更小、独立的子任务的问题。通过合理配置 ForkJoinPool，选择合适的任务粒度，并避免阻塞操作，我们可以编写出高效的并行程序，充分利用多核 CPU 资源。

ForkJoinPool 的工作窃取：平衡线程池负载的算法细节

发表回复 取消回复

发表回复取消回复