C++技能系列( 9 ) - 如何实现线程池【详解】

news2024/11/16 2:26:58

系列文章目录

C++高性能优化编程系列
深入理解软件架构设计系列
高级C++并发线程编程
C++技能系列

期待你的关注哦！！！
在这里插入图片描述

现在的一切都是为将来的梦想编织翅膀，让梦想在现实中展翅高飞。
Now everything is for the future of dream weaving wings, let the dream fly in reality.

如何实现线程池

系列文章目录
一、要实现高效的线程池，可以考虑以下几点
二、实现线程池可以按照以下步骤进行
三、简单的C++线程池代码示例

C++线程池推荐一款基于boost编写的源码库：http://threadpool.sourceforge.net

一、要实现高效的线程池，可以考虑以下几点

控制线程数量： 线程池的大小应该根据系统资源状况和任务量来设置。 太少的线程会导致任务被阻塞，太多的线程则会消耗过多的系统资源。可以使用固定大小的线程池、可缓存的线程池或定时器线程池等方式来进行控制。
任务队列管理： 线程池应该有一个任务队列，用于存放等待执行的任务。 可以使用有界队列或无界队列来管理任务。有界队列可以控制任务的数量，而无界队列则可以接受任意数量的任务，但可能会导致内存溢出。
线程调度策略： 线程池应该有一个合适的线程调度策略，如先进先出、优先级等。可以使用线程池的预定义实现，或自定义实现。
线程错误处理： 线程池应该有一个错误处理机制，用于捕获线程执行过程中可能出现的异常，避免导致整个线程池崩溃。 可以使用try-catch语句或其他异常处理机制来处理异常。
监控和调优： 线程池应该有一个监控和调优机制，用于实时监控线程池的状态和性能，并进行相应的调整。 可以使用监控工具、性能分析工具等来进行监控和调优。

通过合理的设置线程池的大小、任务队列的管理、线程调度策略、错误处理机制和监控调优，可以实现高效的线程池，提高程序的并发性能和资源利用率。

二、实现线程池可以按照以下步骤进行

（1）确定线程池的基本参数： 包括线程池大小、任务队列大小、拒绝策略等。可以根据实际需求来设置这些参数。

（2）创建一个任务队列： 用于存放待执行的任务。可以使用队列数据结构，如ArrayBlockingQueue、LinkedBlockingQueue等。

（3）创建线程池类： 定义一个线程池类，包括线程池的初始化、提交任务、执行任务、关闭等方法。可以使用ThreadPoolExecutor类来实现线程池。

（4）初始化线程池： 在线程池类中，提供一个初始化方法，该方法会根据线程池大小创建固定数量的线程，并将它们放入空闲线程池中。

（5）提交任务： 线程池类提供一个提交任务的方法，用于向任务队列中添加待执行的任务。可以通过调用线程池类的execute方法来实现任务的提交。

（6）执行任务： 线程池会自动从任务队列中获取任务，并将其分配给空闲的线程来执行。任务执行完成后线程会返回到线程池中，等待下一个任务。

（7）关闭线程池： 线程池类提供一个关闭方法，用于停止线程池的运行。线程池在关闭时会等待所有任务执行完毕，然后终止所有线程。

（7）错误处理和监控： 可以在线程池中添加错误处理逻辑，捕获任务执行过程中的异常，避免线程池崩溃。同时，可以添加监控机制，实时监控线程池的状态和性能。

根据上述步骤，可以自定义一个线程池类，实现线程池的功能。在实际使用时，根据具体需求来设置线程池的参数和调优线程池的性能。

三、简单的C++线程池代码示例

该示例中的ThreadPool类实现了一个简单的线程池，包括线程的创建、任务的提交、执行和线程池的关闭等功能。在主函数中使用线程池提交了10个任务，每个任务输出自己的编号和执行它的线程ID。在执行完所有任务之后，程序等待2秒后退出。

请注意，该示例代码只是一个搞笑的演示，可能不具备线程安全和实际应用的一些重要细节，请勿用于实际生产环境。在实际使用线程池时，需要考虑更多的线程同步、任务拆分和异常处理等问题。

#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <functional>

class ThreadPool {
public:
    ThreadPool(int numThreads) : stop(false) {
        for (int i = 0; i < numThreads; ++i) {
            threads.emplace_back(std::thread([this](){
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queueMutex);
                        condition.wait(lock, [this]{ return stop || !tasks.empty(); });
                        if (stop && tasks.empty()) {
                            return;
                        }
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            }));
        }
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        condition.notify_all();
        for (auto& thread : threads) {
            thread.join();
        }
    }

    template <typename FuncType>
    void submit(FuncType f) {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            tasks.emplace([f]() { f(); });
        }
        condition.notify_one();
    }

private:
    std::vector<std::thread> threads;
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop;
};

int main() {
    ThreadPool pool(4);
    for (int i = 0; i < 10; ++i) {
        pool.submit([i]() {
            std::cout << "Task " << i << " executed by thread " << std::this_thread::get_id() << std::endl;
        });
    }
    std::this_thread::sleep_for(std::chrono::seconds(2)); // 等待所有任务执行完
    return 0;
}