自定义线程池性能劣于单线程的原因排查及性能优化

我最近在手写一个Rust线程池，但发现用它处理任务居然比单线程还慢，完全搞不懂问题出在哪。下面是我的线程池实现和用来测试的Criterion基准代码，想请大家帮忙分析下问题根源，以及怎么优化性能。

我的线程池实现 #

use std::sync::atomic::{AtomicBool, AtomicU32};
use std::sync::{
    Arc, atomic::{AtomicUsize, Ordering},
};
use std::thread;
use concurrent_queue::ConcurrentQueue;

pub struct TaskInfo {
    should_stop: bool,
}

pub struct TaskBatchInfo {
    pub task_index: usize,
    pub current_task_count: Arc<AtomicUsize>,
}

pub struct Task {
    func: Box<dyn FnMut() -> Option<TaskInfo> + Send + Sync + 'static>,
}

impl Task {
    fn new(func: impl FnMut() -> Option<TaskInfo> + Send + Sync + Clone + 'static) -> Self {
        Self {
            func: Box::new(func),
        }
    }

    fn run(mut self) -> Option<TaskInfo> {
        (self.func)()
    }
}

pub struct ThreadPool {
    queue: Arc<ConcurrentQueue<Task>>,
    pub should_stop: Arc<AtomicBool>,
    concurrency: usize,
}

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        Self {
            queue: Arc::new(ConcurrentQueue::unbounded()),
            should_stop: Arc::new(AtomicBool::new(false)),
            concurrency: size,
        }
    }

    pub fn start(&self) {
        let queue = self.queue.clone();
        let concurrency = self.concurrency;
        let should_stop = self.should_stop.clone();

        // Spawn the entire thread pool loop on a new thread
        thread::spawn(move || {
            let mut handles = Vec::new();
            for _task_id in 0..concurrency {
                let _queue = queue.clone();
                let _should_stop = should_stop.clone();
                let handle = thread::spawn(move || {
                    loop {
                        if _should_stop.load(Ordering::Relaxed) {
                            break;
                        }
                        let task = _queue.pop();
                        if let Ok(task) = task {
                            let info = task.run();
                            if let Some(info) = info {
                                if info.should_stop {
                                    _should_stop.store(true, Ordering::Relaxed);
                                    break;
                                }
                            }
                        } else {
                            thread::sleep(std::time::Duration::from_micros(500));
                        }
                    }
                });
                handles.push(handle);
            }

            for handle in handles {
                handle.join().unwrap();
            }
        });
    }

    pub fn add_task<F>(&self, task: F)
    where
        F: FnMut() -> Option<TaskInfo> + Send + Sync + Clone + 'static,
    {
        let _ = self.queue.push(Task::new(task));
    }

    pub fn task_batch<F, T>(pool: Arc<Self>, task_count: usize, task: F, termination: T)
    where
        F: FnMut(TaskBatchInfo) -> Option<TaskInfo> + Send + Sync + Clone + 'static,
        T: FnMut() + Send + Sync + Clone + 'static,
    {
        let task_counter = Arc::new(AtomicUsize::new(task_count));
        Self::inner_task_batch(pool, task, task_count, termination, task_counter)
    }

    pub fn task_batch_with_barrier<F, T, C>(
        pool: Arc<Self>,
        task_count: usize,
        mut task: F,
        mut termination: T,
        context: C,
    ) where
        F: FnMut(TaskBatchInfo, C) + Send + Sync + Clone + 'static,
        T: FnMut() + Send + Sync + Clone + 'static,
        C: Clone + Send + Sync + 'static,
    {
        let wait_flag = Arc::new(AtomicU32::new(0));
        let _wait_flag = wait_flag.clone();
        let context = Arc::new(context);

        ThreadPool::task_batch(
            pool.clone(),
            task_count,
            move |info| {
                task(info, context.clone().as_ref().clone());
                None
            },
            move || {
                termination();
                _wait_flag.store(1, Ordering::Relaxed);
                atomic_wait::wake_all(_wait_flag.as_ref());
            },
        );

        atomic_wait::wait(&wait_flag, 0);
    }

    pub fn task_batch_with_barrier_contextless<F>(
        pool: Arc<Self>,
        task_count: usize,
        mut task: F,
    ) where
        F: FnMut(TaskBatchInfo) + Send + Sync + Clone + 'static,
    {
        let wait_flag = Arc::new(AtomicU32::new(0));
        let _wait_flag = wait_flag.clone();

        ThreadPool::task_batch(
            pool.clone(),
            task_count,
            move |info| {
                task(info);
                None
            },
            move || {
                _wait_flag.store(1, Ordering::Relaxed);
                atomic_wait::wake_all(_wait_flag.as_ref());
            },
        );

        atomic_wait::wait(&wait_flag, 0);
    }

    fn inner_task_batch<F, T>(
        pool: Arc<Self>,
        task: F,
        mut task_count: usize,
        termination: T,
        current_task_counter: Arc<AtomicUsize>,
    ) where
        F: FnMut(TaskBatchInfo) -> Option<TaskInfo> + Send + Sync + Clone + 'static,
        T: FnMut() + Send + Sync + Clone + 'static,
    {
        let task_chunk_size = (task_count / pool.as_ref().concurrency).max(1);
        let quotient = task_count / task_chunk_size;
        let remainder = task_count % task_chunk_size;

        for i in 0..quotient + (remainder > 0usize) as usize {
            let mut _task = task.clone();
            let mut _termination = termination.clone();
            let mut _current_task_counter = current_task_counter.clone();
            let _pool = pool.clone();

            pool.add_task(Box::new(move || {
                for j in i * task_chunk_size..((i+1) * task_chunk_size).min(task_count) {
                    _task(TaskBatchInfo {
                        task_index: j,
                        current_task_count: _current_task_counter.clone(),
                    });

                    let val = _current_task_counter.fetch_sub(1, Ordering::Relaxed);
                    if val == 1 {
                        _termination()
                    }
                }
                None
            }));
        }
    }
}

基准测试代码 #

use std::hint::black_box;
use std::sync::{Arc, atomic::Ordering};
use atomic_float::AtomicF64;
use criterion::{Criterion, criterion_group, criterion_main};
use thread_pool::*;

fn st_addition(input: &mut Vec<f64>) {
    for v in input {
        *v += 1.0;
    }
}

fn mt_addition(thread_pool: Arc<ThreadPool>, input: Arc<Vec<AtomicF64>>) {
    // 这里的实现应该是调用线程池的task_batch来处理每个AtomicF64的加法
}

// 基准测试组定义（省略部分细节）
fn bench_addition(c: &mut Criterion) {
    // 初始化测试数据
    let mut st_data: Vec<f64> = (0..1000000).map(|_| 0.0).collect();
    let mt_data: Arc<Vec<AtomicF64>> = Arc::new((0..1000000).map(|_| AtomicF64::new(0.0)).collect());
    let pool = Arc::new(ThreadPool::new(4));
    pool.start();

    c.bench_function("single_thread_addition", |b| b.iter(|| st_addition(black_box(&mut st_data))));
    c.bench_function("multi_thread_addition", |b| b.iter(|| mt_addition(pool.clone(), mt_data.clone())));
}

criterion_group!(benches, bench_addition);
criterion_main!(benches);

性能劣于单线程的核心原因分析 #

我梳理了几个可能导致性能问题的点，大家可以参考：

• 低效的任务轮询机制：worker线程在任务队列空的时候，用thread::sleep(500微秒)来等待，这种方式要么会错过任务到来的时机（睡眠期间有任务进来没法立刻处理），要么频繁唤醒空转，带来不必要的CPU开销。而且ConcurrentQueue本身是非阻塞的，没有提供阻塞等待的接口，导致worker只能不断轮询。

• 任务调度与克隆开销过高：

  • Task要求闭包实现Clone，每次添加任务都会克隆闭包，对于高频任务来说，这部分克隆开销会被放大；
  • inner_task_batch里每次循环都会克隆task、termination、current_task_counter等Arc对象，虽然Arc克隆是轻量的，但累积起来也是不小的开销；
  • 任务块拆分的粒度可能不合理：如果任务块太小，调度的开销会超过并行带来的收益；如果太大，又没法充分利用多核CPU的并行能力。

• 原子操作的额外开销：基准测试里多线程版本用了AtomicF64，而单线程用普通f64，原子操作的内存屏障开销远大于普通内存操作，这直接导致多线程版本的基准测试先天就比单线程慢，不一定是线程池的问题。

• 线程池启动的冗余开销：start方法里把整个worker线程的创建逻辑包在一个额外的线程里，这会导致线程池启动有延迟，而且worker线程在空闲时不会自动退出，一直占用CPU资源。

针对性的性能优化建议 #

针对上面的问题，我整理了几个优化方向：

• 替换为阻塞式任务队列：把concurrent_queue换成crossbeam-channel，它支持阻塞式的recv操作，worker线程在没有任务时会阻塞休眠，有任务进来时立刻被唤醒，彻底消除轮询和睡眠的开销。

• 优化任务粒度与克隆开销：

  • 去掉Task对闭包Clone的约束，改用Arc包裹闭包，避免每次添加任务都克隆闭包；
  • 在inner_task_batch里，减少不必要的Arc克隆，比如current_task_counter可以在任务块里复用同一个引用，不用每次循环都克隆；
  • 根据任务类型调整task_chunk_size：对于计算密集型任务，让每个worker处理的任务块足够大（比如每个worker处理几万条数据），抵消调度开销；对于IO密集型任务，任务粒度可以适当缩小。

• 调整基准测试的公平性：让单线程版本也使用AtomicF64，或者多线程版本改用普通f64（配合分片处理，避免数据竞争），这样对比出来的结果才是线程池本身的性能差异。

• 优化线程池的生命周期管理：

  • 去掉start方法里包裹的额外线程，直接在start里创建worker线程；
  • 给worker线程添加空闲超时逻辑，比如空闲超过几秒就自动退出，避免占用CPU资源；
  • 用parking_lot的条件变量配合队列，实现更高效的阻塞等待。

内容来源于stack exchange