【Linux】系统编程简单线程池(C++)

喜欢ヅ旅行 2024-03-02 10:31 42阅读 0赞

**目录**

【1】线程池概念

【1.1】线程池

【1.2】线程池的应用场景

【1.3】线程池的种类

【1.4】线程池示例

【2】线程池代码

--------------------

## 【1】线程池概念 ##

### 【1.1】线程池 ###

一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

### 【1.2】线程池的应用场景 ###

*  需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
 *  对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
 *  接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限， 出现错误.

### 【1.3】线程池的种类 ###

*  半同步/半异步模式
 *  领导者/跟随着模式

### 【1.4】线程池示例 ###

*  创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执行任务对象中的任务接口。

## 【2】线程池代码 ##

**【Makefile文件】**

# 创建变量关联关系
    cc=c++
    standard=-std=c++11
    linkLib=-l pthread
    
    # 创建编译文件依赖关系
    myThreadPool:ThreadPool.cc
        $(cc) -o $@ $^ $(standard) $(linkLib)  
    
    # 创建删除命令
    .PHONY:clean
    clean:
        rm -f myThreadPool

**【Task.hpp文件】**

#pragma once
    #include <iostream>
    #include <string>
    #include <functional>
    
    /* 计算任务 */
    class CalTask
    {
    public:
        using func_t = std::function<int(int, int, char)>;
    
    public:
        /* 构造函数 */
        CalTask() {}
    
        /* 构造函数 */
        CalTask(int x, int y, char op, func_t func)
            : _x(x), _y(y), _op(op), _callBalk(func)
        {
        }
    
    public:
        /* 仿函数 */
        std::string operator()()
        {
            int result = _callBalk(_x, _y, _op);
    
            char buffer[64];
            snprintf(buffer, sizeof(buffer), "%d %c %d = %d", _x, _op, _y, result);
            return buffer;
        }
    
    public:
        /* 返回打印公式 */
        std::string ToTaskString()
        {
            char buffer[64];
            snprintf(buffer, sizeof(buffer), "%d %c %d = ？", _x, _op, _y);
            return buffer;
        }
    
    private:
        int _x;
        int _y;
        char _op;
        func_t _callBalk;
    };
    
    /* 执行计算的方法 */
    int MyCalculate(int x, int y, char op)
    {
        int result = 0;
        switch (op)
        {
        case '+':
            result = x + y;
            break;
    
        case '-':
            result = x - y;
            break;
    
        case '*':
            result = x * y;
            break;
    
        case '/':
        {
            if (y == 0)
            {
                std::cerr << "div zero error！" << std::endl;
                result = -1;
            }
            else
            {
                result = x / y;
            }
            break;
        }
    
        case '%':
        {
            if (y == 0)
            {
                std::cerr << "mod zero error！" << std::endl;
                result = -1;
            }
            else
            {
                result = x % y;
            }
            break;
        }
    
        default:
            break;
        }
    
        return result;
    }
    
    /* 保存任务 */
    class SaveTask
    {
    public:
        using func_t = std::function<void(const std::string &)>;
    
    public:
        /* 构造函数 */
        SaveTask() {}
        /* 构造函数 */
        SaveTask(const std::string &message, func_t func)
            : _message(message), _callBalk(func)
        {
        }
    
    public:
        /* 仿函数 */
        void operator()()
        {
            _callBalk(_message);
        }
    
    private:
        std::string _message;
        func_t _callBalk;
    };
    
    /* 保存方法 */
    void Save(const std::string& massage){
        std::string target = "./log.txt";
        FILE *fp = fopen(target.c_str(), "a+");
        if(fp == NULL){
            std::cerr << "fopen fail!" << std::endl;
            return;
        }
    
        fputs(massage.c_str(), fp);
        fputs("\n", fp);
        fclose(fp);
    }

**【Thread.hpp文件】**

#pragma once
    #include <iostream>
    #include <functional>
    #include <cstdio>
    #include <cassert>
    
    namespace ThreadNs
    {
        /* 线程连接上下文 */
        class Thread;
        class ConnectText
        {
        public:
            /* 构造函数 */
            ConnectText() : _this(nullptr), _args(nullptr) {}
            /* 析构函数 */
            ~ConnectText() {}
    
        public:
            Thread *_this;
            void *_args;
        };
    
    
        /* 线程类封装 */
        class Thread
        {
        public:
            using func_t = std::function<void *(void *)>; // // 从定义类似函数指针类型：返回值是：void*  参数是：void*
    
        public:
            /* 构造函数 */
            Thread(const int number = 0)
            {
                // 创建线程的名称
                char buffer[64];
                snprintf(buffer, sizeof(buffer), "Thread-%d", number);
                _tName = buffer;
            }
    
            /* 析构函数 */
            ~Thread() {}
            
        public:
            /* 线程启动 */
            void Start(const func_t& func, void *args = nullptr)
            {
                // 建立关系
                _func = func;
                _args = args;
    
                // 创建线程后，启动
                ConnectText *cnt = new ConnectText();
                cnt->_this = this;
                cnt->_args = _args;
    
                int n = pthread_create(&_tid, nullptr, StartRoutine, (void *)cnt);
                assert(n == 0); (void)n;       
                // 编译debug的方式时assert是存在的，release方式assert是不存在的，到时n就是定义了，但是没有被使用的变量。
                // 在有的编译器下会有warning。
            }
    
            /* 线程等待 */
            void Join()
            {
                int n = pthread_join(_tid, nullptr);
                assert(n == 0);
                (void)n;
            }
    
        public:
            /* 获取线程名字 */
            std::string GetThreadName()
            {
                return _tName;
            }
    
        private:
            /* 线程函数 */
            static void *StartRoutine(void *args)
            { // 在类内创建线程，想让线程执行对应的方法，需要将方法设置称为static
                ConnectText *cnt = static_cast<ConnectText *>(args);
                void *exRet = cnt->_this->RoutineRun(cnt->_args);
     
                delete cnt;
                return exRet;
            }
    
            void *RoutineRun(void *args)
            {
                return _func(args);
            }
    
        private:
            pthread_t _tid;     // 线程id
            std::string _tName; // 线程名称
            func_t _func;       // 线程函数
            void *_args;        // 线程参数
        };
    }

**【ThreadPool.hpp文件】**

#pragma once
    #include <iostream>
    #include <vector>
    #include <queue>
    #include <pthread.h>
    #include "Thread.hpp"
    #include "LockGuard.hpp"
    
    /* 引用命名空间 */
    using namespace ThreadNs;
    
    /* 线程上下文数据 */
    template <class T>
    class ThreadPool;
    
    template <class T>
    class ThreadData
    {
    public:
        ThreadData(ThreadPool<T> *tp, const std::string &n)
            : _threadPool(tp), _name(n)
        {}
    
    public:
        ThreadPool<T> *_threadPool;
        std::string _name;
    };
    
    
    
    /* 线程池封装 */
    const int g_num = 5;    // 设置线程数
    template <class T>
    class ThreadPool
    {
    public:
        /* 构造函数 */
        ThreadPool(const int &num = g_num)
            : _threadNum(num)
        {
            // 初始化线程锁
            pthread_mutex_init(&_mutex, nullptr);
            // 初始化线程条件变量
            pthread_cond_init(&_cond, nullptr);
    
            // 将每个线程地址Push到线程地址容器中
            for (int i = 0; i < _threadNum; i++)
            {
                _threads.push_back(new Thread(i + 1));
            }
        }
    
        /* 析构函数 */
        ~ThreadPool()
        {
            // 释放线程锁
            pthread_mutex_destroy(&_mutex);
            // 释放线程条件变量
            pthread_cond_destroy(&_cond);
    
            // 遍历释放
            for (auto &t : _threads)
            {
                delete t;
            }
        }
    
    public:
        /* 开启线程池 */
        void Run()
        {
            // 启动线程
            for (const auto &t : _threads)
            {
                ThreadData<T> *td = new ThreadData<T>(this, t->GetThreadName());
                t->Start(HandlerTask, td);
                std::cout << t->GetThreadName() << ": Start..." << std::endl;
            }
    
            // // 阻塞式等待回收线程
            // for (const auto &t : _threads)
            // {
            //     t->Join();
            //     std::cout << t->GetThreadName() << ": Recycle..." << std::endl;
            // }
        }
    
    public:
        /* 获取锁 */
        pthread_mutex_t *Mutex() { return &_mutex; }
        /* 线程加锁 */
        void Lock() { pthread_mutex_lock(&_mutex); }
        /* 线程解锁 */
        void UnLock() { pthread_mutex_unlock(&_mutex); }
        /* 线程等待 */
        void ThreadWait() { pthread_cond_wait(&_cond, &_mutex); }
        /* 判断是否有任务 */
        bool IsEmpty() { return _taskQueue.empty(); }
    
        /* 新增任务 */
        void Push(const T &in)
        {
            // 加锁->自动解锁
            LockGuard LockGuard(&_mutex);
            // 新增任务
            _taskQueue.push(in);
            // 环形线程执行
            pthread_cond_signal(&_cond);
        }
    
        /* 执行任务 */
        T Pop()
        {
            // 创建T对象
            T t;
            // 获取栈顶任务
            t = _taskQueue.front();
            _taskQueue.pop();
            // 返回任务
            return t;
        }
    
    private:
        /* 线程池共享线程 */
        static void *HandlerTask(void *args)
        {
            ThreadData<T> *td = static_cast<ThreadData<T> *>(args);
    
            while (true)
            {
                // 创建任务对象
                T t;
    
                {
                    // 加锁->自动解锁
                    LockGuard LockGuard(td->_threadPool->Mutex());
                    // 如果没有任务，等待任务
                    if (td->_threadPool->IsEmpty())
                    {
                        td->_threadPool->ThreadWait();
                    }
                    t = td->_threadPool->Pop();
                    // 解锁
                    td->_threadPool->UnLock();
                }
    
                // 打印信息
                std::cout << td->_name << "：承接一个任务[" << t.ToTaskString() << "],任务的执行结果是[" << t() << "]" << std::endl;
            }
            return nullptr;
        }
    
    private:
        int _threadNum;                 // 线程数量
        std::vector<Thread *> _threads; // 存放线程地址的容器
        std::queue<T> _taskQueue;       // 存放线程任务的队列
        pthread_mutex_t _mutex;         // 线程锁
        pthread_cond_t _cond;           // 线程条件变量
    };

**【ThreadPool.cc文件】**

#include <iostream>
    #include <memory>
    #include <unistd.h>
    #include "Task.hpp"
    #include "ThreadPool.hpp"
    using namespace std;
    
    int main() 
    {
        // 智能指针管理
        unique_ptr<ThreadPool<CalTask>> tP(new ThreadPool<CalTask>());
        // 启动线程池
        tP->Run();
    
        // 手动派发任务
        int x;
        int y;
        char op;
        while (1)
        {
            std::cout << "请输入数据1# ";
            std::cin >> x;
            std::cout << "请输入数据2# ";
            std::cin >> y;
            std::cout << "请输入要进行的运算# ";
            std::cin >> op;
            CalTask t(x, y, op, MyCalculate);
            tP->Push(t);
            sleep(1);
        }
        
        return 0;
    }

**【代码测试结果】**

![60a1c26b5e634ecbb48eca748ac7553f.png][]

[60a1c26b5e634ecbb48eca748ac7553f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/4708bd5e2d314c4e9e9835b53d0e45b2.png