C++多线程控制问询：如何保持≤5个线程并发处理25个元素

核心思路 #

我们需要一种「动态监控线程状态，一旦有线程结束就立即补充新线程」的机制，本质上就是实现一个固定大小的简易线程池，或者用标准库工具动态管理并发任务的生命周期。

下面给两种可行的实现方案：

方案1：用std::async + std::future动态管理任务 #

利用std::async创建异步任务并返回std::future，我们维护一个最多容纳5个future的容器，每次循环时先清理已经完成的任务，再添加新任务，确保同时运行的任务数始终≤5。

#include <vector>
#include <thread>
#include <future>
#include <algorithm>

void func(const std::vector<std::string>& input, int idx) {
    // 你的线程处理逻辑，比如：
    // std::cout << "Processing element " << idx << ": " << input[idx] << std::endl;
}

void test(std::vector<std::string>& input) {
    std::vector<std::future<void>> futures;
    futures.reserve(5); // 预分配空间优化性能

    for (int i = 0; i < 25; ++i) {
        // 先清理已经完成的任务，释放线程槽
        futures.erase(
            std::remove_if(futures.begin(), futures.end(),
                [](std::future<void>& f) { return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready; }),
            futures.end()
        );

        // 当有空闲槽时，启动新任务
        if (futures.size() < 5) {
            // 注意：如果input不需要拷贝，用std::ref(input)传递引用，避免不必要的拷贝开销
            futures.emplace_back(std::async(std::launch::async, func, std::cref(input), i));
        }
    }

    // 等待所有剩余任务完成
    for (auto& f : futures) {
        f.get();
    }
}

方案说明 #

• std::launch::async确保任务在新线程执行（而不是延迟到get()/wait()时才执行）
• wait_for(std::chrono::seconds(0))是非阻塞检查任务是否完成，不会阻塞主线程
• 每次循环先清理已完成的future，再添加新任务，严格控制并发数≤5

方案2：实现固定大小的线程池（更高效的复用方案） #

如果任务数量较多，频繁创建销毁线程会有额外开销，这时候预先创建5个工作线程，让它们从任务队列中取任务执行，是更优的选择：

#include <vector>
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>

void func(const std::vector<std::string>& input, int idx) {
    // 你的线程处理逻辑
}

class FixedThreadPool {
public:
    FixedThreadPool(size_t num_threads) {
        for (size_t i = 0; i < num_threads; ++i) {
            threads.emplace_back([this]() {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(mtx);
                        cv.wait(lock, [this]() { return stop || !tasks.empty(); });
                        if (stop && tasks.empty()) return;
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    ~FixedThreadPool() {
        {
            std::unique_lock<std::mutex> lock(mtx);
            stop = true;
        }
        cv.notify_all();
        for (auto& th : threads) {
            th.join();
        }
    }

    void enqueue(std::function<void()> task) {
        {
            std::unique_lock<std::mutex> lock(mtx);
            tasks.emplace(std::move(task));
        }
        cv.notify_one();
    }

private:
    std::vector<std::thread> threads;
    std::queue<std::function<void()>> tasks;
    std::mutex mtx;
    std::condition_variable cv;
    bool stop = false;
};

void test(std::vector<std::string>& input) {
    FixedThreadPool pool(5); // 固定5个工作线程

    for (int i = 0; i < 25; ++i) {
        // 捕获当前i的副本，避免闭包引用问题
        pool.enqueue([&input, i]() {
            func(input, i);
        });
    }

    // 线程池会在析构时自动等待所有任务完成
}

方案说明 #

• 预先创建5个线程，持续从任务队列中取任务执行，避免频繁创建销毁线程的开销
• 用条件变量std::condition_variable实现线程的等待/唤醒机制，当有新任务时唤醒一个空闲线程
• 线程池的析构函数会自动等待所有任务完成并join线程，无需手动管理

对比你的原实现，这两种方案都能保证任意时刻运行的线程数不超过5，并且只要有线程完成就立即启动新任务，最大化利用CPU资源。

内容的提问来源于stack exchange，提问作者Karthik K M