C++如何实现高性能的多线程工作窃取队列来优化并发性能

多线程并发编程中，任务分配不均是常见性能瓶颈，工作窃取队列通过允许空闲线程窃取忙碌线程的待执行任务，实现负载均衡，提升整体执行效率。本文介绍C++实现高性能工作窃取队列的完整方案。

工作窃取队列的核心设计思路

工作窃取队列通常采用双端队列结构，每个工作线程维护自己的任务队列：线程自身从队列的一端（通常是尾部）推送和弹出任务，其他空闲线程从队列的另一端（头部）窃取任务，这样能减少不同线程操作同一端带来的竞争。

为了实现高性能，需要尽可能采用无锁设计，避免互斥锁带来的开销，同时要考虑缓存友好性，减少伪共享问题。核心操作包括三个：

• 本地线程推送任务到队列尾部
• 本地线程从队列尾部弹出任务执行
• 其他线程从队列头部窃取任务

核心数据结构定义

首先定义任务类型和队列的基础结构，任务用函数对象表示，队列内部用环形数组存储任务，配合头尾索引实现双端操作。

#include <atomic>
#include <vector>
#include <functional>
#include <thread>
#include <cassert>

// 任务类型，无参数无返回值的函数对象
using Task = std::function<void()>;

class WorkStealingQueue {
private:
    // 环形数组存储任务，大小固定为2的幂次方便取模
    std::vector<Task> tasks;
    // 队列容量，必须是2的幂次
    size_t capacity;
    // 掩码，用于快速取模 capacity - 1
    size_t mask;
    // 头部索引，窃取线程操作的一端，使用原子变量保证可见性
    std::atomic<size_t> head;
    // 尾部索引，本地线程操作的一端，使用原子变量保证可见性
    std::atomic<size_t> tail;
    // 缓存行填充，避免head和tail伪共享
    char padding[64 - sizeof(std::atomic<size_t>)];

public:
    // 构造函数，初始化容量为2的幂次
    explicit WorkStealingQueue(size_t cap = 1024) : capacity(1) {
        // 确保容量是2的幂次
        while (capacity < cap) {
            capacity <<= 1;
        }
        mask = capacity - 1;
        tasks.resize(capacity);
        head.store(0, std::memory_order_relaxed);
        tail.store(0, std::memory_order_relaxed);
    }

    // 禁止拷贝构造和赋值
    WorkStealingQueue(const WorkStealingQueue&) = delete;
    WorkStealingQueue& operator=(const WorkStealingQueue&) = delete;
};

核心操作实现

本地推送任务

本地线程向队列尾部推送任务，只需要修改tail索引，不需要考虑其他线程的窃取操作，因为窃取操作只修改head索引，两者操作不同的原子变量，减少竞争。

    // 本地线程推送任务到队列尾部，返回是否成功
    bool push(Task task) {
        size_t t = tail.load(std::memory_order_relaxed);
        size_t h = head.load(std::memory_order_relaxed);
        // 队列已满，当前简单返回失败，实际可扩容
        if (t - h >= capacity) {
            return false;
        }
        tasks[t & mask] = std::move(task);
        // 使用release语义，保证任务写入对其他线程可见
        tail.store(t + 1, std::memory_order_release);
        return true;
    }

本地弹出任务

本地线程从队列尾部弹出任务，先读取tail和head，判断队列是否为空，不为空则修改tail索引，使用acquire语义保证读取到正确的任务。

    // 本地线程从队列尾部弹出任务，返回是否成功
    bool pop(Task& task) {
        size_t t = tail.load(std::memory_order_relaxed);
        size_t h = head.load(std::memory_order_relaxed);
        // 队列为空
        if (t == h) {
            return false;
        }
        // 先减tail，再读取任务，避免和窃取操作冲突
        t -= 1;
        tail.store(t, std::memory_order_relaxed);
        // 使用acquire语义，保证之前push的任务写入可见
        std::atomic_thread_fence(std::memory_order_acquire);
        task = std::move(tasks[t & mask]);
        // 再次检查是否有窃取操作发生，避免任务被重复取走
        h = head.load(std::memory_order_relaxed);
        if (t >= h) {
            return true;
        }
        // 窃取操作已经取走了所有任务，当前弹出无效
        tail.store(h, std::memory_order_relaxed);
        return false;
    }

窃取任务操作

其他线程从队列头部窃取任务，需要读取head和tail，判断队列是否有任务，然后尝试修改head索引，使用CAS操作保证原子性，避免多个窃取线程同时窃取同一个任务。

    // 其他线程从队列头部窃取任务，返回是否成功
    bool steal(Task& task) {
        size_t h = head.load(std::memory_order_relaxed);
        size_t t = tail.load(std::memory_order_acquire);
        // 队列为空
        if (h >= t) {
            return false;
        }
        task = tasks[h & mask];
        // 使用CAS操作修改head，避免多个窃取线程冲突
        if (head.compare_exchange_strong(h, h + 1, std::memory_order_acq_rel, std::memory_order_relaxed)) {
            return true;
        }
        // CAS失败，说明其他线程已经窃取了任务，返回失败
        return false;
    }

完整使用示例

下面是一个简单的使用示例，创建多个工作线程，每个线程维护自己的工作窃取队列，空闲时尝试窃取其他线程的任务。

#include <iostream>
#include <vector>
#include <mutex>

std::mutex cout_mutex;

void test_work_stealing() {
    const int THREAD_NUM = 4;
    const int TASK_NUM = 1000;
    std::vector<std::thread> threads;
    std::vector<WorkStealingQueue> queues(THREAD_NUM);

    // 初始化任务，全部放到第一个线程的队列中
    for (int i = 0; i < TASK_NUM; ++i) {
        int task_id = i;
        queues[0].push([task_id]() {
            std::lock_guard<std::mutex> lock(cout_mutex);
            std::cout << "Task " << task_id << " executed by thread " << std::this_thread::get_id() << std::endl;
        });
    }

    // 启动工作线程
    for (int i = 0; i < THREAD_NUM; ++i) {
        threads.emplace_back([i, &queues]() {
            Task task;
            // 不断尝试从本地队列取任务，取不到就尝试窃取其他线程的任务
            while (true) {
                // 先尝试本地弹出
                if (queues[i].pop(task)) {
                    task();
                    continue;
                }
                // 本地没有任务，尝试窃取其他线程的任务
                bool stolen = false;
                for (int j = 0; j < queues.size(); ++j) {
                    if (j == i) continue;
                    if (queues[j].steal(task)) {
                        task();
                        stolen = true;
                        break;
                    }
                }
                if (!stolen) {
                    // 所有队列都没有任务，退出循环（实际场景可加休眠或退出条件）
                    break;
                }
            }
        });
    }

    // 等待所有线程结束
    for (auto& t : threads) {
        t.join();
    }
}

int main() {
    test_work_stealing();
    return 0;
}

性能优化要点

• 队列容量设置为2的幂次，用位运算代替取模操作，提升计算效率
• 头尾索引使用原子变量，避免互斥锁开销，本地操作和窃取操作操作不同的索引，减少竞争
• 增加缓存行填充，避免head和tail两个原子变量处于同一个缓存行，减少伪共享带来的性能损耗
• 任务推送和弹出使用合适的内存序，在保证正确性的前提下降低内存屏障的开销
• 窃取操作使用CAS而不是互斥锁，多个窃取线程竞争时开销更小

注意事项

工作窃取队列适合任务粒度较小的场景，如果任务执行时间很长，窃取带来的收益会被任务执行时间掩盖。另外，任务对象需要支持移动语义，避免任务拷贝带来的开销。如果队列需要动态扩容，要额外处理扩容时的并发安全问题，或者提前预估任务数量设置合适的初始容量。

C++多线程work_stealing并发调优无锁队列修改时间：2026-07-13 20:54:40