解析 ‘Global vs. Local Run Queues’ 的负载均衡：如何在大规模多核（128+ Cores）场景下减少锁竞争？

各位听众，下午好！非常荣幸今天能与大家共同探讨一个在现代高性能计算领域至关重要的话题：大规模多核系统（特别是128核以上）中的负载均衡挑战，以及如何通过优化运行队列（Run Queues）的设计来有效减少锁竞争。

随着芯片技术的飞速发展，我们正步入一个“超多核”时代。CPU核心数量从几十个跃升到上百个，甚至更多。然而，核心数量的增加并非总是带来线性甚至超线性性能提升。往往，我们发现应用程序的扩展性瓶颈不再是计算能力本身，而是核心之间的数据同步、资源协调以及最常见的——锁竞争。今天，我们将聚焦于操作系统调度器和用户态线程池中运行队列的设计，剖析全局队列与本地队列的优劣，并深入探讨在128+核心的庞然大物中，如何巧妙地设计调度机制以“驯服”锁竞争这头猛兽。

运行队列与调度器的基石

首先，让我们建立一个共同的理解基础。什么是运行队列？简单来说，运行队列是一个数据结构，用于存放那些已经准备好运行但尚未被CPU核心执行的任务（或线程）。调度器是操作系统的“大脑”，它负责从运行队列中选择任务，并将它们分配给空闲的CPU核心执行。在用户态线程池中，这一角色则由线程池管理器扮演。

一个典型的任务调度循环是这样的：

1. 任务提交： 新任务被创建或提交到调度器。
2. 入队： 任务被放入运行队列。
3. 出队与执行： CPU核心（或线程池中的工作线程）从队列中取出任务并执行。
4. 任务完成/阻塞： 任务执行完毕或因等待资源而阻塞。
5. 重新入队： 阻塞的任务在资源就绪后，可能再次被放入运行队列。

运行队列的设计直接影响着调度的效率、公平性以及最重要的——系统的整体吞吐量和延迟。

挑战：128+ 核心的锁竞争噩梦

设想一下，在一个拥有128个甚至更多物理核心的系统上，如果所有核心都尝试访问同一个共享资源，其后果将是灾难性的。在运行队列的场景中，这个共享资源就是队列本身。

1. 单一全局运行队列：简单即是陷阱

最直观、最简单的设计莫过于使用一个单一的全局运行队列。所有任务都提交到这个队列，所有核心都从这个队列中获取任务。

架构示意：

                  +-------------------+
                  | Global Run Queue  |
                  | (e.g., std::deque)|
                  | Protected by      |
                  | std::mutex        |
                  +-------------------+
                       ^         ^
                       |         |
          Dequeue Task |         | Enqueue Task
                       |         |
+---------+      +---------+      +---------+      +---------+
| Core 0  |<---->| Core 1  |<---->| ...     |<---->| Core 127|
+---------+      +---------+      +---------+      +---------+

优点：

• 实现简单： 只需要一个共享队列和一个互斥锁即可。
• 完美负载均衡（理论上）： 只要队列中有任务，任何空闲核心都可以立即获取并执行，不会出现核心空闲而任务堆积的情况。

缺点（在大规模多核场景下是致命的）：

• 极高的锁竞争： 每一个核心在每次尝试获取任务、提交任务或更新队列状态时，都需要获取并释放同一个全局互斥锁。在128个核心同时竞争这把锁时，绝大多数时间将浪费在锁的等待上，而不是实际的计算。
• 缓存颠簸（Cache Line Bouncing）： 互斥锁本身通常占用一个缓存行。当多个核心频繁地竞争同一把锁时，这个缓存行会在不同核心的私有缓存之间来回失效、传递，导致大量的缓存同步开销。此外，队列头部和尾部的数据也可能因此频繁地在不同核心之间移动，进一步加剧缓存颠簸。
• 性能瓶颈： 锁的开销不再是微不足道的，它将成为系统性能的绝对瓶颈。随着核心数量的增加，系统吞吐量甚至可能不升反降。

代码示例（概念性）：

#include <deque>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <vector>
#include <iostream>
#include <atomic>

// 模拟一个任务
struct Task {
    int id;
    std::string name;
    // ... 其他任务数据
};

// 全局运行队列
std::deque<Task> globalRunQueue;
std::mutex globalMutex;
std::condition_variable globalCv;
std::atomic<bool> stopWorkers(false);

// 工作线程函数 (模拟一个CPU核心)
void workerThread(int coreId) {
    std::cout << "Core " << coreId << " started." << std::endl;
    while (!stopWorkers.load(std::memory_order_acquire)) {
        Task task;
        bool taskFound = false;

        { // 尝试从全局队列获取任务
            std::unique_lock<std::mutex> lock(globalMutex);
            globalCv.wait(lock, [&]{ 
                return !globalRunQueue.empty() || stopWorkers.load(std::memory_order_acquire); 
            });

            if (stopWorkers.load(std::memory_order_acquire)) {
                break;
            }

            if (!globalRunQueue.empty()) {
                task = globalRunQueue.front();
                globalRunQueue.pop_front();
                taskFound = true;
            }
        } // 锁在这里释放

        if (taskFound) {
            // std::cout << "Core " << coreId << " executing task " << task.id << std::endl;
            // 模拟任务执行
            std::this_thread::sleep_for(std::chrono::microseconds(10)); 
        } else {
            // std::cout << "Core " << coreId << " idle, waiting..." << std::endl;
            // 短暂休眠以避免自旋
            std::this_thread::sleep_for(std::chrono::microseconds(1));
        }
    }
    std::cout << "Core " << coreId << " stopped." << std::endl;
}

// 生产者函数 (模拟任务提交者)
void producerThread(int numTasks) {
    for (int i = 0; i < numTasks; ++i) {
        Task t = {i, "Task_" + std::to_string(i)};
        {
            std::lock_guard<std::mutex> lock(globalMutex);
            globalRunQueue.push_back(t);
        }
        globalCv.notify_one(); // 通知一个等待的worker
        // std::this_thread::sleep_for(std::chrono::microseconds(1)); // 模拟任务生成间隔
    }
    std::cout << "Producer finished generating " << numTasks << " tasks." << std::endl;
}

// int main() {
//     const int NUM_CORES = 4; // 示例用少量核心，方便观察
//     const int NUM_TASKS = 1000;

//     std::vector<std::thread> workers;
//     for (int i = 0; i < NUM_CORES; ++i) {
//         workers.emplace_back(workerThread, i);
//     }

//     std::thread producer(producerThread, NUM_TASKS);

//     producer.join();

//     // 等待所有任务被消费
//     while (true) {
//         std::unique_lock<std::mutex> lock(globalMutex);
//         if (globalRunQueue.empty()) {
//             break;
//         }
//         // 等待队列清空，或者设置一个超时防止死锁
//         // 实际场景中，生产者停止后，worker会消费完所有任务
//         // 这里简单等待，如果任务量大，需要更复杂的退出机制
//         std::this_thread::sleep_for(std::chrono::milliseconds(10)); 
//     }
//     std::cout << "All tasks consumed." << std::endl;

//     stopWorkers.store(true, std::memory_order_release);
//     globalCv.notify_all(); // 唤醒所有等待的worker使其退出

//     for (auto& w : workers) {
//         w.join();
//     }

//     std::cout << "Simulation finished." << std::endl;
//     return 0;
// }

分析： 即使是4个核心，频繁的锁竞争也会导致性能下降。当核心数达到128甚至更多时，globalMutex将成为一个高度争用的热点，导致大量线程阻塞和上下文切换，极大地降低CPU的有效利用率。实际的性能数据会显示，随着核心数的增加，程序的吞吐量在达到某个峰值后，会因为锁竞争而迅速下降。

2. 本地运行队列：去中心化的力量

为了解决全局队列的锁竞争问题，自然而然地会想到将队列分散开来。每个CPU核心（或每个工作线程）维护一个自己的本地运行队列。

架构示意：

+---------+       +---------+       +---------+       +---------+
| Core 0  |       | Core 1  |       | ...     |       | Core 127|
| +-------+ |     | +-------+ |     |         |     | +-------+ |
| | Local | |     | | Local | |     |         |     | | Local | |
| | Queue | |     | | Queue | |     |         |     | | Queue | |
| +-------+ |     | +-------+ |     |         |     | +-------+ |
+-----------+     +-----------+     +-----------+     +-----------+

优点：

• 显著减少锁竞争： 大多数时候，核心只需要访问自己的本地队列，这几乎是无锁的（或只涉及轻量级的原子操作）。任务的入队和出队操作，如果没有外部干预，将非常高效。
• 优异的缓存局部性： 任务及其相关数据倾向于在同一个核心上处理，这意味着CPU缓存（L1/L2/L3）的命中率会更高，减少了内存访问延迟。
• 更好的扩展性： 核心之间操作互相独立的队列，理论上可以无限扩展，不会因为队列操作而出现中心瓶颈。

缺点：

• 负载不均衡： 这是本地队列最核心的问题。如果某些核心的任务量很大，而另一些核心的任务量很少甚至空闲，就会出现“忙者愈忙，闲者愈闲”的局面。系统整体的CPU利用率会很低，任务的完成时间也会不可预测。
• 任务迁移开销： 为了解决负载不均衡，需要引入任务迁移机制。将任务从一个核心的队列移动到另一个核心的队列，会带来额外的开销（数据复制、缓存失效等）。
• 实现复杂性增加： 需要额外的机制来检测并解决负载不均衡问题。

代码示例（概念性）：

#include <deque>
#include <mutex>
#include <thread>
#include <vector>
#include <iostream>
#include <atomic>
#include <random> // For random task distribution

// 模拟一个任务
struct Task {
    int id;
    std::string name;
    // ... 其他任务数据
};

// 每个核心一个本地运行队列
struct CoreLocalQueue {
    std::deque<Task> queue;
    std::mutex mutex; // 保护本地队列
    std::condition_variable cv;
};

std::vector<CoreLocalQueue> localRunQueues;
std::atomic<bool> stopWorkers(false);
std::atomic<int> totalTasksProcessed(0);

// 工作线程函数
void workerThreadLocal(int coreId) {
    std::cout << "Core " << coreId << " started." << std::endl;
    CoreLocalQueue& myQueue = localRunQueues[coreId];

    while (!stopWorkers.load(std::memory_order_acquire)) {
        Task task;
        bool taskFound = false;

        { // 尝试从自己的本地队列获取任务
            std::unique_lock<std::mutex> lock(myQueue.mutex);
            myQueue.cv.wait_for(lock, std::chrono::milliseconds(10), 
                                  [&]{ return !myQueue.queue.empty() || stopWorkers.load(std::memory_order_acquire); });

            if (stopWorkers.load(std::memory_order_acquire)) {
                break;
            }

            if (!myQueue.queue.empty()) {
                task = myQueue.queue.front();
                myQueue.queue.pop_front();
                taskFound = true;
            }
        } // 本地锁在这里释放

        if (taskFound) {
            // std::cout << "Core " << coreId << " executing local task " << task.id << std::endl;
            std::this_thread::sleep_for(std::chrono::microseconds(10));
            totalTasksProcessed.fetch_add(1, std::memory_order_relaxed);
        } else {
            // std::cout << "Core " << coreId << " local queue empty, might try to steal..." << std::endl;
            // 短暂休眠以避免自旋，或在此处引入工作窃取逻辑
            std::this_thread::sleep_for(std::chrono::microseconds(1));
        }
    }
    std::cout << "Core " << coreId << " stopped." << std::endl;
}

// 生产者函数 (模拟任务提交者，随机分发任务)
void producerThreadLocal(int numTasks, int numCores) {
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<> distrib(0, numCores - 1);

    for (int i = 0; i < numTasks; ++i) {
        Task t = {i, "Task_" + std::to_string(i)};
        int targetCore = distrib(gen); // 随机选择一个核心

        CoreLocalQueue& targetQueue = localRunQueues[targetCore];
        {
            std::lock_guard<std::mutex> lock(targetQueue.mutex);
            targetQueue.queue.push_back(t);
        }
        targetQueue.cv.notify_one();
        // std::this_thread::sleep_for(std::chrono::microseconds(1));
    }
    std::cout << "Producer finished generating " << numTasks << " tasks." << std::endl;
}

// int main() {
//     const int NUM_CORES = 4;
//     const int NUM_TASKS = 1000;

//     localRunQueues.resize(NUM_CORES);

//     std::vector<std::thread> workers;
//     for (int i = 0; i < NUM_CORES; ++i) {
//         workers.emplace_back(workerThreadLocal, i);
//     }

//     std::thread producer(producerThreadLocal, NUM_TASKS, NUM_CORES);

//     producer.join();

//     // 等待所有任务被消费
//     auto start_time = std::chrono::high_resolution_clock::now();
//     while (totalTasksProcessed.load(std::memory_order_acquire) < NUM_TASKS) {
//         std::this_thread::sleep_for(std::chrono::milliseconds(10));
//         if (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::high_resolution_clock::now() - start_time).count() > 5) {
//             std::cerr << "Timeout waiting for tasks to complete. " << totalTasksProcessed.load() << " / " << NUM_TASKS << " processed." << std::endl;
//             break; // 防止无限等待
//         }
//     }
//     std::cout << "All tasks consumed." << std::endl;

//     stopWorkers.store(true, std::memory_order_release);
//     for(auto& q : localRunQueues) {
//         q.cv.notify_all(); // 唤醒所有等待的worker使其退出
//     }

//     for (auto& w : workers) {
//         w.join();
//     }

//     std::cout << "Simulation finished." << std::endl;
//     return 0;
// }

分析： 在这个本地队列的示例中，我们看到每个核心的队列操作都是独立的，锁竞争大大减少。但如果 producerThreadLocal 不均匀地分发任务，例如只分发给 Core 0，那么 Core 0 会非常忙碌，而其他核心则会空闲。这就是负载不均衡的问题。

3. 混合策略：工作窃取与工作共享

为了兼顾低锁竞争和负载均衡，现代高性能调度器普遍采用混合策略，其中最流行和有效的是“工作窃取”（Work Stealing）。

工作窃取（Work Stealing）

核心思想： 忙碌的生产者将任务加入到其本地队列的“尾部”（底部），工作线程从其本地队列的“头部”（顶部）获取任务。当一个工作线程发现自己的本地队列为空时，它不会坐以待毙，而是会主动去“窃取”其他（通常是随机选择的）工作线程队列中的任务。为了减少窃取时的竞争，窃取操作通常从被窃取队列的“尾部”（底部）进行。

数据结构： 双端队列（Deque）是实现工作窃取最常见的数据结构，因为它允许在两端进行高效的插入和删除操作。

窃取机制：

1. 本地操作： 线程A从其本地队列Q_A的头部取出任务。
2. 窃取操作： 线程A发现Q_A为空，随机选择一个“受害者”线程B，尝试从Q_B的尾部窃取任务。这个窃取操作需要获取Q_B的锁。
3. 冲突缓解： 大多数窃取操作是无竞争的，因为它们发生在不同的队列上。只有当多个窃取者同时瞄准同一个受害者，或者受害者线程自身正在修改队列尾部时，才会发生锁竞争。

优点：

• 优秀的负载均衡： 闲置核心不会浪费计算能力，能够主动参与到任务处理中。
• 低锁竞争： 大多数时候，核心只操作自己的队列，仅在窃取时发生锁竞争，且通常竞争不如全局队列激烈（因为窃取是稀疏事件，且目标分散）。
• 良好的缓存局部性： 任务优先在本地执行，直到本地队列为空才进行窃取。
• 自适应性： 能够很好地适应动态变化的负载模式。

缺点：

• 实现复杂： 需要精心设计的数据结构（如无锁或高效锁的双端队列）和协调机制。
• 窃取开销： 窃取操作本身有开销，包括锁开销、任务迁移开销、缓存失效开销。
• “热点”窃取： 如果少数几个核心长期处于极度繁忙状态，它们可能会成为所有空闲核心的窃取目标，从而再次引入锁竞争。

代码示例（工作窃取简化版）：

#include <deque>
#include <mutex>
#include <thread>
#include <vector>
#include <iostream>
#include <atomic>
#include <random>
#include <chrono>

// 模拟一个任务
struct Task {
    int id;
    std::string name;
    // ... 其他任务数据
};

// 每个核心一个本地运行队列 (使用 std::mutex 简化，实际可能使用更轻量级或无锁的实现)
struct CoreLocalStealingQueue {
    std::deque<Task> queue;
    std::mutex mutex; 
    std::condition_variable cv; // 用于本地等待
    std::atomic<bool> active_tasks_present; // 快速检查是否有任务，避免每次都获取锁
};

std::vector<CoreLocalStealingQueue> localStealingQueues;
std::atomic<bool> stopWorkersStealing(false);
std::atomic<long long> totalTasksProcessedStealing(0);
std::vector<std::thread> workerThreadsStealing; // 存储工作线程

// 工作线程函数 (包含工作窃取逻辑)
void workerThreadStealing(int coreId, int numCores) {
    std::cout << "Core " << coreId << " started." << std::endl;
    CoreLocalStealingQueue& myQueue = localStealingQueues[coreId];
    std::random_device rd;
    std::mt19937 gen(rd() + coreId); // 每个线程使用不同的随机种子

    while (!stopWorkersStealing.load(std::memory_order_acquire)) {
        Task task;
        bool taskFound = false;

        // 1. 尝试从本地队列获取任务 (从头部)
        {
            std::unique_lock<std::mutex> lock(myQueue.mutex);
            // 本地队列为空时，等待一小段时间或直接进入窃取
            myQueue.cv.wait_for(lock, std::chrono::microseconds(50), 
                                  [&]{ return !myQueue.queue.empty() || stopWorkersStealing.load(std::memory_order_acquire); });

            if (stopWorkersStealing.load(std::memory_order_acquire)) {
                break;
            }

            if (!myQueue.queue.empty()) {
                task = myQueue.queue.front();
                myQueue.queue.pop_front();
                taskFound = true;
            }
        } // 本地锁释放

        if (taskFound) {
            // std::cout << "Core " << coreId << " executing local task " << task.id << std::endl;
            std::this_thread::sleep_for(std::chrono::microseconds(10));
            totalTasksProcessedStealing.fetch_add(1, std::memory_order_relaxed);
            continue; // 继续处理下一个本地任务
        }

        // 2. 本地队列为空，尝试窃取任务
        if (!stopWorkersStealing.load(std::memory_order_acquire)) {
            // 随机选择一个受害者队列
            std::uniform_int_distribution<> distrib(0, numCores - 1);
            int victimCore = -1;
            for (int i = 0; i < numCores; ++i) { // 尝试多次，避免总是选到空队列
                victimCore = distrib(gen);
                if (victimCore == coreId) continue; // 不窃取自己

                CoreLocalStealingQueue& victimQueue = localStealingQueues[victimCore];
                // 快速检查 victimQueue 是否可能有任务，避免不必要的锁竞争
                // 实际无锁队列会提供更好的is_empty()方法

                std::unique_lock<std::mutex> victimLock(victimQueue.mutex, std::try_to_lock);
                if (victimLock.owns_lock()) { // 成功获取受害者队列的锁
                    if (!victimQueue.queue.empty()) {
                        task = victimQueue.queue.back(); // 从尾部窃取
                        victimQueue.queue.pop_back();
                        taskFound = true;
                        // std::cout << "Core " << coreId << " STOLE task " << task.id << " from Core " << victimCore << std::endl;
                        break; // 窃取成功，跳出循环
                    }
                }
            }

            if (taskFound) {
                std::this_thread::sleep_for(std::chrono::microseconds(10));
                totalTasksProcessedStealing.fetch_add(1, std::memory_order_relaxed);
            } else {
                // 如果没有窃取到任务，短暂休眠避免忙等
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
        }
    }
    std::cout << "Core " << coreId << " stopped." << std::endl;
}

// 生产者函数 (将任务均匀分发或集中分发，以便观察窃取效果)
void producerThreadStealing(int numTasks, int numCores) {
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<> distrib(0, numCores - 1);

    for (int i = 0; i < numTasks; ++i) {
        Task t = {i, "Task_" + std::to_string(i)};
        int targetCore = distrib(gen); // 随机分发
        // int targetCore = 0; // 或者集中分发给 Core 0，观察窃取效果更明显

        CoreLocalStealingQueue& targetQueue = localStealingQueues[targetCore];
        {
            std::lock_guard<std::mutex> lock(targetQueue.mutex);
            targetQueue.queue.push_back(t);
        }
        targetQueue.cv.notify_one();
        // std::this_thread::sleep_for(std::chrono::microseconds(1));
    }
    std::cout << "Producer finished generating " << numTasks << " tasks." << std::endl;
}

// int main() {
//     const int NUM_CORES = 8; // 增加核心数以观察窃取效果
//     const int NUM_TASKS = 50000;

//     localStealingQueues.resize(NUM_CORES);

//     for (int i = 0; i < NUM_CORES; ++i) {
//         workerThreadsStealing.emplace_back(workerThreadStealing, i, NUM_CORES);
//     }

//     std::thread producer(producerThreadStealing, NUM_TASKS, NUM_CORES);
//     producer.join();

//     // 等待所有任务被消费
//     auto start_time = std::chrono::high_resolution_clock::now();
//     while (totalTasksProcessedStealing.load(std::memory_order_acquire) < NUM_TASKS) {
//         std::this_thread::sleep_for(std::milliseconds(50));
//         if (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::high_resolution_clock::now() - start_time).count() > 10) {
//             std::cerr << "Timeout waiting for tasks to complete. " << totalTasksProcessedStealing.load() << " / " << NUM_TASKS << " processed." << std::endl;
//             break; 
//         }
//     }
//     std::cout << "All tasks (" << totalTasksProcessedStealing.load() << ") consumed." << std::endl;

//     stopWorkersStealing.store(true, std::memory_order_release);
//     for(auto& q : localStealingQueues) {
//         q.cv.notify_all(); // 唤醒所有等待的worker使其退出
//     }

//     for (auto& w : workerThreadsStealing) {
//         w.join();
//     }

//     std::cout << "Simulation finished." << std::endl;
//     return 0;
// }

分析： 上述代码演示了工作窃取的基本逻辑。当一个核心本地队列为空时，它会尝试从其他核心的队列尾部“窃取”任务。std::try_to_lock在这里很重要，它避免了在无法立即获取锁时线程被阻塞，而是继续尝试其他队列或短暂休眠。在128+核心的场景下，为了进一步优化，通常会采用无锁（lock-free）或等待无关（wait-free）的双端队列，以彻底消除窃取时的互斥锁竞争。例如，Linux内核的CFS（Completely Fair Scheduler）就采用了工作窃取机制，但其队列实现更为复杂和高效。

工作共享（Work Sharing）

核心思想： 与工作窃取相反，工作共享是主动行为。当一个核心的队列变得过于繁忙（任务数量超过某个阈值）时，它会主动将一部分任务推送到其他空闲或负载较轻的核心的队列中。

优点：

• 更可控的负载均衡： 调度器可以根据全局负载信息进行更精细的控制。
• 减少窃取时的开销： 避免了空闲核心盲目地尝试窃取。

缺点：

• 需要全局状态或协调： 核心需要了解其他核心的负载情况，这可能需要一个中心化的负载监控机制，再次引入竞争或同步开销。
• 实现复杂： 如何判断“过于繁忙”和“负载较轻”？如何选择目标核心？如何避免任务在核心之间来回“乒乓”？这些都是挑战。

在实际系统中，工作窃取往往更受欢迎，因为它是一种去中心化的、响应式的负载均衡策略，通常能更好地扩展。工作共享更多地作为一种辅助手段或在特定场景下使用。

4. 进一步减少锁竞争的策略

在工作窃取的基础上，针对128+核心的场景，我们还需要考虑更多高级技术来进一步减少锁竞争和优化性能。

a. 无锁（Lock-Free）/等待无关（Wait-Free）队列

这是减少锁竞争的终极目标。通过使用原子操作（如CAS, Compare-And-Swap）而非互斥锁来设计队列，可以完全消除线程阻塞，从而避免上下文切换和锁开销。

• MPMC (Multi-Producer, Multi-Consumer) 队列： 多个生产者和多个消费者都可以并发地操作队列。实现难度极高，需要精通内存模型和原子操作。
• SPSC (Single-Producer, Single-Consumer) 队列： 一个生产者和一个消费者操作队列。相对简单，但不能用于工作窃取（因为窃取意味着多个消费者）。

无锁队列的挑战：

• ABA 问题： CAS操作只检查值是否相同，但不检查期间是否有其他修改。
• 内存序（Memory Ordering）： 需要谨慎使用 std::memory_order 来保证操作的可见性和顺序。
• 复杂性： 即使是专家也常常难以正确实现无锁数据结构。通常建议使用经过严格测试的库（如folly::MPMCQueue）。

概念性无锁队列片段（非常简化，仅展示CAS思路）：

// 假设有一个基于数组的环形队列
template<typename T>
class LockFreeQueue {
private:
    std::atomic<size_t> head;
    std::atomic<size_t> tail;
    std::vector<T> buffer;
    size_t capacity;

public:
    LockFreeQueue(size_t cap) : head(0), tail(0), capacity(cap) {
        buffer.resize(cap);
    }

    bool enqueue(const T& value) {
        size_t currentTail = tail.load(std::memory_order_relaxed);
        size_t nextTail = (currentTail + 1) % capacity;

        if (nextTail == head.load(std::memory_order_acquire)) { // 队列已满
            return false;
        }

        buffer[currentTail] = value;
        tail.store(nextTail, std::memory_order_release);
        return true;
    }

    bool dequeue(T& value) {
        size_t currentHead = head.load(std::memory_order_relaxed);
        if (currentHead == tail.load(std::memory_order_acquire)) { // 队列为空
            return false;
        }

        value = buffer[currentHead];
        head.store((currentHead + 1) % capacity, std::memory_order_release);
        return true;
    }
};

注意： 上述 LockFreeQueue 仅为概念性示例，在多生产者多消费者场景下，简单的 head 和 tail 原子操作不足以保证正确性，还需要复杂的CAS循环和内存屏障来处理竞争条件和ABA问题。实际生产中，应使用成熟的无锁队列实现。

b. NUMA 感知（NUMA-Awareness）

在128+核心的系统上，很可能涉及多个NUMA节点。跨NUMA节点的内存访问（尤其是缓存行传输）会比本地NUMA节点内慢得多。

• NUMA 拓扑感知： 调度器应了解CPU和内存的NUMA拓扑结构。
• 本地化策略： 优先将任务调度到其数据所在的NUMA节点上的核心。
• NUMA-Aware 工作窃取： 当一个核心需要窃取任务时，它首先尝试从同一NUMA节点内的其他核心队列窃取。如果所有本地NUMA节点队列都为空，才考虑跨NUMA节点窃取。这可以显著减少跨NUMA访问的延迟和带宽消耗。

示例：NUMA-Aware 窃取策略

// 假设每个NUMA节点有自己的CoreLocalStealingQueue组
std::vector<std::vector<CoreLocalStealingQueue>> numaNodeQueues; 

void workerThreadNUMAAwareStealing(int coreId, int myNumaNodeId, const std::vector<int>& coresInMyNumaNode) {
    // ... (本地队列处理)

    // 1. 尝试从本NUMA节点内的其他核心窃取
    for (int victimCore : coresInMyNumaNode) {
        if (victimCore == coreId) continue;
        // 尝试从 localStealingQueues[victimCore] 窃取
        // ...
        if (taskFound) return;
    }

    // 2. 如果本NUMA节点内无任务，才尝试跨NUMA节点窃取（随机选择或根据负载选择）
    // ...
}

c. 批量操作（Batching）

每次获取锁只处理一个任务的开销是巨大的。如果能够一次性获取锁，并从队列中取出或放入一批任务，就可以分摊锁的开销。

• 批量出队： 工作线程一次性从队列中取出N个任务，然后一次性释放锁，再逐个执行这些任务。
• 批量入队： 生产者将多个任务打包成一个批次，一次性将整个批次放入队列。

示例：批量出队

// 在 workerThreadStealing 中获取任务的部分
std::vector<Task> tasksToProcess;
const int BATCH_SIZE = 8; // 每次尝试获取8个任务

{
    std::unique_lock<std::mutex> lock(myQueue.mutex);
    myQueue.cv.wait_for(lock, std::chrono::microseconds(50), 
                          [&]{ return !myQueue.queue.empty() || stopWorkersStealing.load(std::memory_order_acquire); });

    if (stopWorkersStealing.load(std::memory_order_acquire)) {
        break;
    }

    for (int i = 0; i < BATCH_SIZE && !myQueue.queue.empty(); ++i) {
        tasksToProcess.push_back(myQueue.queue.front());
        myQueue.queue.pop_front();
    }
} // 锁在这里释放

for (const auto& task : tasksToProcess) {
    // 执行任务
    std::this_thread::sleep_for(std::chrono::microseconds(10));
    totalTasksProcessedStealing.fetch_add(1, std::memory_order_relaxed);
}

通过批量操作，虽然锁的持有时间可能略微增加，但锁的获取/释放频率大大降低，整体的锁竞争和缓存颠簸会减少。

d. 层次化队列（Hierarchical Queues）

对于超大规模系统，可以考虑多级运行队列。

• 核心本地队列： 如前所述，每个核心一个。
• NUMA 节点本地队列： 每个NUMA节点有一个共享队列，用于节点内部的负载均衡或作为核心窃取失败后的备用队列。
• 全局队列（可选，慎用）： 作为最底层的备用队列，在极端不均衡或所有本地队列都空时使用，但通常应避免。

这种层次结构可以在不同粒度上进行负载均衡，并且在越高的层次（越全局），队列的访问频率和锁竞争应越低。

e. 线程亲和性（Thread Affinity）与任务绑定

对于某些对缓存局部性要求极高的任务，可以将其绑定到特定的核心或核心组。这减少了任务在核心之间迁移的可能性，但需要谨慎使用，以免破坏整体负载均衡。

f. 优先级队列（Priority Queues）

如果系统中有不同优先级的任务，运行队列也可以是优先级队列。但这会增加队列操作的复杂性和潜在的锁竞争，因为维护优先级通常比简单FIFO队列更复杂。通常，在大规模吞吐量系统中，简单FIFO队列配合高效的负载均衡足以满足大部分需求。

总结比较表格

特性/策略	单一全局运行队列	本地运行队列（无窃取）	工作窃取（本地队列+窃取）
实现难度	简单	简单	中等偏高（尤其无锁实现）
锁竞争	极高（致命瓶颈）	极低（仅本地队列锁）	低（主要在窃取时，且可优化）
缓存局部性	差（数据频繁跨核心移动）	极好（任务倾向于本地执行）	好（优先本地执行，窃取时可能损失）
负载均衡	完美（理论上）	极差（可能严重不均）	优秀（自适应，动态调整）
扩展性	差（核心数越多越差）	极好（操作独立）	优秀（去中心化，可扩展）
任务延迟	不可预测（受锁竞争影响大）	某些任务可能等待过久	相对稳定，空闲核心能及时处理
128+ 核心适用性	不适用	仅适用于特定严格分区场景	推荐（配合无锁/NUMA-Aware/批量）

性能监控与调优

无论选择何种策略，在大规模多核系统上进行调度器设计，都离不开严谨的性能监控和调优。

• CPU 利用率： 关注每个核心的利用率，识别空闲或过载的核心。
• 上下文切换次数： 高频的上下文切换通常是锁竞争或任务切换开销大的信号。
• 缓存命中率/缺失率： 尤其是L3缓存和跨NUMA访问的统计。
• 锁等待时间/竞争次数： 使用性能分析工具（如 perf, VTune, SystemTap）来识别热点锁。
• 任务延迟/吞吐量： 这是最终衡量调度器性能的关键指标。

通过这些指标，我们可以不断迭代和优化调度策略，确保系统在高并发、高负载下依然能够高效运行。

总结思考

在大规模多核系统中，运行队列的设计从根本上决定了系统的可扩展性和性能。从单一全局队列的简单到工作窃取机制的精妙，我们看到的是一个从中心化到去中心化、从粗粒度锁到细粒度或无锁演进的过程。针对128+核心的挑战，核心的策略在于：最大限度地减少共享状态和锁竞争，优先保持数据局部性，并通过智能的负载均衡机制（如NUMA感知的无锁工作窃取）来确保所有计算资源都能被高效利用。这需要深入理解底层硬件架构、内存模型以及并发编程的精髓，并辅以持续的性能分析和调优。

今天的探讨希望能为大家在设计大规模多核系统中的负载均衡策略时提供一些有益的启示和思考。谢谢大家！