【Linux】线程终结篇：线程池以及线程池的实现-Toy模板网

这篇具有很好参考价值的文章主要介绍了【Linux】线程终结篇：线程池以及线程池的实现。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

linux线程完结

文章目录

前言
一、线程池的实现
二、了解性知识
- 1.其他常见的各种锁
- 2.读者写者问题
总结

前言

什么是线程池呢？

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

线程池的应用场景：

1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB 服务器完成网页请求这样的务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为 Telnet 会话时间比线程的创建时间大多了。

2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。

3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.

一、线程池的实现

线程池的使用会用到我们之前自己设计的lockguard，我会将代码放在最后。

下面我们创建一个线程池文件和main.cc文件，然后先写一个线程池的框架：

const int gnum = 5;

template <class T>
class ThreadPool
{
public:
    ThreadPool(const int &num = gnum)
        : _num(num)
    {

    }
    ~ThreadPool()
    {
        
    }

private:
    int _num;
    vector<pthread_t *> _threads;
    queue<T> _task_queue;
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;
};

我们的线程池是用vector进行管理的，里面存放每一个线程对象的地址。然后我们还需要一个队列，这个队列存放的是线程要执行的任务，我们的目的是让多个线程去竞争任务。既然是多线程那么必须要有一把锁来防止线程安全问题，而线程每次去任务队列拿任务，如果有任务就拿没有任务就阻塞在队列中，所以我们还需要一个条件变量。接下来我们解释构造函数，我们在构造函数中需要确定线程池中要创建多少个线程，所以需要定义一个全局变量gnum来当缺省参数。

    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 pthread_t);
        }
    }
    static void* handerTask(void* args)
    {

    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
        for (auto& t: _threads)
        {
            delete t;
        }
    }

我们在构造函数初始化的时候先将锁和条件变量初始化了，然后挨个给vector中的线程指针开一个线程的空间，并且让它们去执行handerTask方法。在析构函数中我们需要将锁和条件变量释放掉，然后挨个将vector中每个线程指针指向的资源释放。

    static void* handerTask(void* args)
    {
        while (true)
        {
            sleep(1);
            cout<<"thread "<<pthread_self()<<"run....."<<std::endl;
        }
        return nullptr;
    }

然后我们写一个线程启动方法让线程启动（也就是创建线程）。

    void start()
    {
        for (const auto& t: _threads)
        {
            pthread_create(t,nullptr,handerTask,nullptr);
            cout<<pthread_self()<<"start...."<<endl;
        }
    }

有了线程启动后我们就可以先写main函数测试一下我们的代码有没有问题：

#include "ThreadPool.hpp"
#include <memory>

int main()
{
    std::unique_ptr<ThreadPool<int>> tp(new ThreadPool<int>());
    tp->start();
    while (1)
    {
        sleep(1);
    }
    return 0;
}

在这里我们用了智能指针来管理线程池，然后将线程池启动，下面我们运行起来：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

可以看到我们前面写的代码是没有问题的，下面我们继续编写线程池执行任务的代码：

我们的任务队列要面对多个线程来抢任务的情景，所以任务队列必须要加锁。

    static void* handerTask(void* args)
    {
        while (true)
        {
            pthread_mutex_lock(&_mutex);
            while (_task_queue.empty())
            {
                pthread_cond_wait(&_cond,&_mutex);
            }
            //获取任务队列中的任务
            T t = _task_queue.front();
            //处理任务
            t();
            pthread_mutex_unlock(&_mutex);
        }
        return nullptr;
    }

加锁后我们还有判断任务队列是否为空，如果为空则需要去条件变量中等待。如果不为空我们就可以获取条件变量中的任务，然后利用仿函数去处理任务。

既然有了任务队列那么我们肯定是要向任务队列中添加任务的，所以我们再写一个Push接口：

    void Push(const T& in)
    {
        pthread_mutex_lock(&_mutex);
        _task_queue.push(in);
        pthread_cond_signal(&_cond);
        pthread_mutex_unlock(&_mutex);
    }

对于添加任务我们首先要做的还是先加锁，然后将任务添加进去，一旦添加了任务我们就可以唤醒阻塞在条件变量中的线程。

实际上当我们写完代码才发现handerTask这个接口是有问题的，因为我们定义的是静态成员函数，在静态成员函数内部是不可以使用普通成员变量的，这个时候我们就需要封装一批接口来供这个方法使用：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

    void lockQueue()
    {
        pthread_mutex_lock(&_mutex);
    }
    void unlockQueue()
    {
        pthread_mutex_unlock(&_mutex);
    }
    void condwaitQueue()
    {
        pthread_cond_wait(&_cond,&_mutex);
    }
    bool IsQueueEmpty()
    {
        return _task_queue.empty();
    }
    T popQueue()
    {
        T t = _task_queue.front();
        _task_queue.pop();
        return t;
    }

下面我们用这些接口对handerTask进行修改：

    static void* handerTask(void* args)
    {
        ThreadPool<T>* threadpool = static_cast<ThreadPool<T>*>(args);
        while (true)
        {
            threadpool->lockQueue();
            while (threadpool->IsQueueEmpty())
            {
                threadpool->condwaitQueue();
            }
            //获取任务队列中的任务
            T t = threadpool->popQueue();
            //处理任务
            //t();
            threadpool->unlockQueue();
            t();
        }
        return nullptr;
    }

这样我们就完成了handerTask方法的设计，但是我们处理任务的代码是有问题的，我们设计的是多线程的模型，让多个线程共同去抢任务执行，如果我们将处理任务的方法放到锁中，那么这个处理任务的过程就变成了串行的，就不符合我们的预期了，所以我们处理任务的过程一定是在解锁后，接下来我们把前一篇文章用到的Task任务导入：

int main()
{
    std::unique_ptr<ThreadPool<Task>> tp(new ThreadPool<Task>());
    tp->start();
    int x ,y;
    char op;
    while (1)
    {
        cout<<"请输入数据1#: ";
        cin>>x;
        cout<<"请输入数据2#: ";
        cin>>y;
        cout<<"请输入你要进行的运算#: ";
        cin>>op;
        Task t(x,y,op);
        tp->Push(t);
        sleep(1);
    }
    return 0;
}

当然为了让运行的结果更容易观察，我们在hander方法中让线程处理完成任务后打印一下结果：

    static void* handerTask(void* args)
    {
        ThreadPool<T>* threadpool = static_cast<ThreadPool<T>*>(args);
        while (true)
        {
            threadpool->lockQueue();
            while (threadpool->IsQueueEmpty())
            {
                threadpool->condwaitQueue();
            }
            //获取任务队列中的任务
            T t = threadpool->popQueue();
            //处理任务
           // t();   注意处理任务不能放在加锁过程中，否则就变成串行的了
            threadpool->unlockQueue();
            t();
            cout<<t.formatArg()<<"?   "<<"的运算的结果为："<<t.formatRes()<<endl;
        }
        return nullptr;
    }

下面我们看看运行后的效果：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

可以看到是没有问题的。

下面我们将线程池中所有加锁的东西都用我们自己写的lockguard做一下整合：

我们先写一个接口用来拿到锁：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

然后我们就可以修改一下hander方法：

 static void* handerTask(void* args)
    {
        ThreadPool<T>* threadpool = static_cast<ThreadPool<T>*>(args);
        while (true)
        {
            T t;
            {
                LockGuard(threadpool->getMutex());
                while (threadpool->IsQueueEmpty())
                {
                    threadpool->condwaitQueue();
                }
                // 获取任务队列中的任务
                t = threadpool->popQueue();
            }
            t();
            cout<<t.formatArg()<<"?   "<<"的运算的结果为："<<t.formatRes()<<endl;
        }
        return nullptr;
    }

当然我们也可以不要用匿名对象，直接定义一下，不然生命周期会有问题：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

    void Push(const T& in)
    {
        LockGuard lock(&_mutex);
        _task_queue.push(in);
        pthread_cond_signal(&_cond); 
    }

修改后我们重新运行一下：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

可以看到是没有问题的。

下面我们将这个线程池改为单例模式：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

首先将构造函数设置为私有，然后将拷贝构造和赋值删除。

下面我们定义一个静态的线程池指针：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

下面我们再设计一个启动单例模式的方法：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

注意我们的启动方法一般都是静态的，因为我们要求这个方法只属于这个类。

然后main函数中原来的指针就变成了用类名直接调用静态方法：

int main()
{
    ThreadPool<Task>::getInstance()->start();
    int x ,y;
    char op;
    while (1)
    {
        cout<<"请输入数据1#: ";
        cin>>x;
        cout<<"请输入数据2#: ";
        cin>>y;
        cout<<"请输入你要进行的运算#: ";
        cin>>op;
        Task t(x,y,op);
        ThreadPool<Task>::getInstance()->Push(t);
        sleep(1);
    }
    return 0;
}

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

然后我们运行起来也是正常的，当然我们获取单列对象的方法一个线程是没有问题的，但是当多线程并发访问就会有问题，就有可能存在多次new对象的情况，所以我们在创建单例模式的时候加一把锁：

【Linux】线程终结篇：线程池以及线程池的实现,linux,c++,后端,linux,线程池,读写锁,STL,线程安全

我们这里加的锁是C++库中的，为什么不用之前的那个锁呢？因为我们的获取单例对象的函数是静态的，并且这个函数不需要传this参数，不像我们之前hander方法需要传this参数正好可以用我们之前定义的Linux中的锁。

    static ThreadPool<T>* getInstance()
    {
        if (_tp == nullptr)
        {
            _mtx.lock();
            if (_tp == nullptr)
            {
                _tp = new ThreadPool<T>();
            }
            _mtx.unlock();
        }
        return _tp;
    }

加锁后我们的静态成员函数的代码如上，至于为什么要判断两次，这是因为我们只有第一次获取单例对象的时候才需要加锁，如果已经有对象了我们还要加锁解锁那么就会浪费资源，所以我们判断两次保证只有第一次进来创建单例对象的时候才加锁。这样就完成了一个简单的单例模式的设定。

STL中的指针是否是线程安全的？

不是 . 原因是, STL 的设计初衷是将性能挖掘到极致 , 而一旦涉及到加锁保证线程安全 , 会对性能造成巨大的影响 . 而且对于不同的容器, 加锁方式的不同 , 性能可能也不同 ( 例如 hash 表的锁表和锁桶 ).

因此 STL 默认不是线程安全 . 如果需要在多线程环境下使用 , 往往需要调用者自行保证线程安全。

智能指针是否是线程安全的 ?

对于 unique_ptr, 由于只是在当前代码块范围内生效 , 因此不涉及线程安全问题 .

对于 shared_ptr, 多个对象需要共用一个引用计数变量 , 所以会存在线程安全问题 . 但是标准库实现的时候考虑到了这个问题, 基于原子操作 (CAS) 的方式保证 shared_ptr 能够高效 , 原子的操作引用计数 .

其他常见的各种锁

悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行 锁等），当其他线程想要访问数据时，被阻塞挂起。

乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前， 会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。

CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不相 等则失败，失败则重试，一般是一个自旋的过程，即不断重试。

自旋锁，公平锁，非公平锁等（自旋锁就是每隔一段时间轮询式的看自己是否是否需要加锁解锁）

读者写者问题

读写锁

在编写多线程的时候，有一种情况是十分常见的。那就是，有些公共数据修改的机会比较少。相比较改写，它们读的机会反而高的多。通常而言，在读的过程中，往往伴随着查找的操作，中间耗时很长。给这种代码段加锁，会极大地降低我们程序的效率。那么有没有一种方法，可以专门处理这种多读少写的情况呢？有，那就是读写锁。

读写锁接口

设置读写优先：

int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
/*
pref 共有 3 种选择
PTHREAD_RWLOCK_PREFER_READER_NP (默认设置) 读者优先，可能会导致写者饥饿情况
PTHREAD_RWLOCK_PREFER_WRITER_NP 写者优先，目前有 BUG，导致表现行为和
PTHREAD_RWLOCK_PREFER_READER_NP 一致
PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP 写者优先，但写者不能递归加锁
*/

初始化：

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,const pthread_rwlockattr_t
*restrict attr);

销毁：

int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

加锁和解锁：

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);