Linux如何设计一个线程池

在设计线程池之前，我们可以对线程进行简单的封装这样子在线程池中就可以少一点调用接口，就像搭积木一样，一层一层的搭上去

#pragma once
 
#include <iostream>
#include <pthread.h>
#include <string>
#include <functional>
#include <cassert>
 
// 上下文
 
namespace ThreadNs
{
    const int num = 1024;
    typedef std::function<void *(void *)> func_t;
    class Thread
    {
    private:
        // 在类内创建线程，想让线程执行对应的方法，需要将方法设置为static
        static void *start_routine(void *argv) // 类内成员，有缺省参数！this指针
        {
            // 静态方法不能调用成员方法和成员变量
            // 将成员设置成为静态的不推荐，友元也可以
            Thread *_this = static_cast<Thread *>(argv);
            return _this->calback();
        }
 
    public:
        Thread(const Thread &t)
            : _func(t._func), _args(t._args), _name(t._name)
        {
        }
        Thread() 
        {
            // _name = "thread-";
            // _name += std::to_string(number);
 
            char namebuffer[num];
            snprintf(namebuffer, sizeof namebuffer, "thread - %d", Threadnum++);
            _name = namebuffer;
        }
 
        void start(func_t func, void *args = nullptr)
        {
            _func = func;
            _args = args;
            int n = pthread_create(&_tid, nullptr, start_routine, this);
            assert(n == 0);
            (void)n; // 有些编译器会报warning
        }
        void *calback() { return _func(_args);}
        std::string threadname()
        {
            return _name;
        }
        void join()
        {
            int n = pthread_join(_tid, nullptr);
            assert(n == 0);
            (void)n;
        }
        ~Thread()
        {
            // do nothing
        }
 
    private:
        std::string _name;
        pthread_t _tid;
        func_t _func;
        void *_args;
 
        static int Threadnum;
    };
    int Thread::Threadnum = 1;
} // namespace ThreadNs

这里的命名方式都是使用前_代表类内成员，我们将构造函数的参数设计的简单一点，方便后面实施，然后用_args代表要喂给线程的参数，_func代表回调函数。给每个线程设计一个名字使用_name

线程池里面肯定要有线程，不过在此之前我们可以写一个Task的类方便后面测试，我这里写一个示例

#pragma once
 
#include <iostream>
#include <functional>
 
class Task
{
public:
    using func_t = std::function<int(int,int, char)>;
    // typedef std::function<int(int,int)> func_t;
    Task()
    {}
    Task(int x, int y,char op, func_t func):_x(x),_y(y), _op(op),_callback(func)
    {}
    std::string operator()()
    {
        int result = _callback(_x,_y, _op);
        char buffer[1024];
        snprintf(buffer, sizeof buffer, "%d %c %d = %d", _x, _op, _y, result);
        return buffer;
    }
    std::string toTaskstring()
    {
        char buffer[1024];
        snprintf(buffer, sizeof buffer, "%d %c %d = ?", _x, _op, _y);
        return buffer;        
    }
private:
    int _x;
    int _y;
    char _op;
    func_t _callback;
};
const std::string oper = "+-*/%";
 
int mymath(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;
}
 

这是一个简单的算术任务类。

接下来负责写线程池，线程池中肯定要有一系列线程，还有一个队列用于拿任务，还要有一个互斥锁和条件变量。当你完成了积木的底层以后上层就很容易了

#pragma once
 
#include "Mutex.hpp"
#include "Thread.hpp"
#include "Task.hpp"
#include <vector>
#include <queue>
#include <mutex>
#include <pthread.h>
#include <unistd.h>
 
using namespace ThreadNs;
const int gnum = 5;
template <class T>
class ThreadPool;
template <class T>
class ThreadData
{
public:
    ThreadPool<T> *_tp;
    std::string _name;
 
public:
    ThreadData(ThreadPool<T> *tp, const std::string &name) : _tp(tp), _name(name) {}
};
template <class T>
class ThreadPool
{
private:
    static void *handlerTask(void *args)
    {
        ThreadData<T> *td = (ThreadData<T> *)args;
        // ThreadPool<T>* tp = static_cast<ThreadPool<T>*> (args);
        while (1)
        {
            // sleep(1);
            // std::cout << "thread " << pthread_self() << " run ..." << std::endl;
            // td->_tp->lockQueue();
            T t;
            {
 
                LockGuard lockgurad(td->_tp->mutex());
                while (td->_tp->isEmptyQueue())
                {
                    /* code */
                    td->_tp->threadwait();
                }
                t = td->_tp->Pop();
            }
            // td->_tp->unlockQueue();
            std::cout << td->_name << "处理完了任务：" << t.toTaskstring() << t() << std::endl;
        }
        delete td;
        return nullptr;
    }
 
public:
    void lockQueue() { pthread_mutex_lock(&_mutex); }
    void unlockQueue() { pthread_mutex_unlock(&_mutex); }
    bool isEmptyQueue() { return _task_queue.empty(); }
    void threadwait() { pthread_cond_wait(&_cond, &_mutex); }
    pthread_mutex_t *mutex() { return &_mutex; }
    T Pop()
    {
        T t = _task_queue.front();
        _task_queue.pop();
        return t;
    }
 
    ThreadPool(const int &num = gnum) : _num(num)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
        for (int i = 0; i < _num; i++)
        {
            _threads.push_back(new Thread());
        }
    }
 
    ThreadPool(const ThreadPool &) = delete;
    ThreadPool operator=(const ThreadPool &) = delete;
 
public:
    void run()
    {
        for (const auto &t : _threads)
        {
            ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
            t->start(handlerTask, td);
            std::cout << t->threadname() << "start ..." << std::endl;
        }
    }
    void push(const T &in)
    {
        // pthread_mutex_lock(&_mutex);
        LockGuard lockgurad(&_mutex);
        _task_queue.push(in);
 
        pthread_cond_signal(&_cond);
        // pthread_mutex_unlock(&_mutex);
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
        for (const auto &t : _threads)
            delete t;
    }
    // 最好加static
    static ThreadPool<T> *getInstance()
    {
        // LockGuard(&_mutex);
        if (nullptr == tp)
        {
            _singlock.lock();
            if (nullptr == tp)
            {
                tp = new ThreadPool<Task>();
            }
            _singlock.unlock();
        }
 
        return tp;
    }
 
private:
    int _num;
    std::vector<Thread *> _threads;
    std::queue<T> _task_queue;
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;
 
    static ThreadPool<T> *tp;
    static std::mutex _singlock;
};
 
template <class T>
ThreadPool<T> *ThreadPool<T>::tp = nullptr;
 
template <class T>
std::mutex ThreadPool<T>::_singlock;