Linux之 线程池 | 单例模式的线程安全问题 | 其他锁

一、线程池

1、线程池

线程池是一种线程使用模式。线程池里面可以维护一些线程。

为什么要有线程池？

因为在我们使用线程去处理各种任务的时候，尤其是一些执行时间短的任务，我们必须要先对线程进行创建然后再进行任务处理，最后再销毁线程，效率是比较低的。而且有的时候线程过多会带来调度开销，进而影响缓存局部性和整体性能。

于是，我们可以通过线程池预先创建出一批线程，线程池维护着这些线程，线程等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。

线程池不仅能够保证内核的充分利用，还能防止过分调度。

2、线程池代码

我们先对线程进行封装：Thread.hpp

#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <pthread.h>

using namespace std;
typedef void *(*fun_t)(void *);

class ThreadData
{
public:
    void *arg_;
    string name_;
};

class Thread
{
public:
    Thread(int num, fun_t callback, void *arg)
        : func_(callback)
    {
        char buffer[64];
        snprintf(buffer, sizeof(buffer), "Thread-%d", num);
        name_ = buffer;
        tdata_.name_ = name_;
        tdata_.arg_ = arg;
    }

    void start()
    {
        pthread_create(&tid_, nullptr, func_, (void *)&tdata_);
    }

    void join()
    {
        pthread_join(tid_, nullptr);
    }

    string &name()
    {
        return name_;
    }

    ~Thread()
    {
    }

private:
    pthread_t tid_;
    string name_;
    fun_t func_;
    ThreadData tdata_;
};

线程池代码：threadPool.hpp：

#pragma once
#include <vector>
#include <queue>
#include "thread.hpp"

#define THREAD_NUM 3

template <class T>
class ThreadPool
{
public:
    bool Empty()
    {
        return task_queue_.empty();
    }

    pthread_mutex_t *getmutex()
    {
        return &lock;
    }

    void wait()
    {
        pthread_cond_wait(&cond, &lock);
    }

    T gettask()
    {
        T t = task_queue_.front();
        task_queue_.pop();
        return t;
    }

public:
    ThreadPool(int num = THREAD_NUM) : num_(num)
    {
        for (int i = 0; i < num_; i++)
        {
            threads_.push_back(new Thread(i, routine, this));
        }
        pthread_mutex_init(&lock, nullptr);
        pthread_cond_init(&cond, nullptr);
    }

    static void *routine(void *arg)
    {
        ThreadData *td = (ThreadData *)arg;
        ThreadPool<T> *tp = (ThreadPool<T> *)td->arg_;
        while (true)
        {
            T task;
            {
                pthread_mutex_lock(tp->getmutex());
                while (tp->Empty())
                    tp->wait();
                task = tp->gettask();
                pthread_mutex_unlock(tp->getmutex());
            }
            cout << "x+y=" << task() << " " << pthread_self() << endl;
        }
    }

    void run()
    {
        for (auto &iter : threads_)
        {
            iter->start();
        }
    }

    void PushTask(const T &task)
    {
        pthread_mutex_lock(&lock);
        task_queue_.push(task);
        pthread_mutex_unlock(&lock);
        pthread_cond_signal(&cond);
    }

    ~ThreadPool()
    {
        for (auto &iter : threads_)
        {
            iter->join();
            delete iter;
        }
        pthread_mutex_destroy(&lock);
        pthread_cond_destroy(&cond);
    }

private:
    vector<Thread *> threads_;
    int num_;
    queue<T> task_queue_;
    pthread_mutex_t lock;
    pthread_cond_t cond;
};

任务：task.hpp：

#pragma once

#include <iostream>
#include <queue>
#include <pthread.h>
#include <unistd.h>

class task
{
public:
    task()
    {
    }
    task(int x, int y)
        : x_(x), y_(y)
    {
    }

    int operator()()
    {
        return x_ + y_;
    }

private:
    int x_;
    int y_;
};

测试代码：test.cc：

#include "threadPool.hpp"
#include "task.hpp"
#include <iostream>
#include <ctime>

int main()
{
    srand((unsigned int)time(nullptr) ^ getpid() ^ 12232);
    ThreadPool<task> *tp = new ThreadPool<task>();
    tp->run();
    while (true)
    {
        int x = rand() % 100 + 1;
        sleep(1);
        int y = rand() % 100 + 1;
        task t(x, y);
        tp->PushTask(t);
        cout << x << "+" << y << "=?" << endl;
    }

    return 0;
}

运行结果：

3、线程池的应用场景

1、需要大量的线程来完成任务，且完成任务的时间比较短。
2、对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
3、接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。

二、单例模式的线程安全问题

1、线程池的单例模式

首先，我们要做的第一件事就是把构造函数私有，再把拷贝构造和赋值运算符重载函数delete：

private:
    ThreadPool(int num = THREAD_NUM) : num_(num)
    {
        for (int i = 0; i < num_; i++)
        {
            threads_.push_back(new Thread(i, routine, this));
        }
        pthread_mutex_init(&lock, nullptr);
        pthread_cond_init(&cond, nullptr);
    }

    ThreadPool(const TreadPool &other) = delete;
    ThreadPool operator=(const TreadPool &other) = delete;

接下来就要在类中定义一个成员变量：静态指针，方便获取单例对象，并在类外初始化：

//线程池中的成员变量
private:
    vector<Thread *> threads_;
    int num_;
    queue<T> task_queue_;
    pthread_mutex_t lock;
    pthread_cond_t cond;

    static ThreadPool<T> *tp;

//在类外初始化
​template <class T>
ThreadPool<T> *ThreadPool<T>::tp = nullptr;

最后我们写一个函数可以获取单例对象，在设置获取单例对象的函数的时候，注意要设置成静态成员函数，因为在获取对象前根本没有对象，无法调用非静态成员函数（无this指针）:

static ThreadPool<T> *getThreadPool()
{
    if (tp == nullptr)
    {
        tp = new ThreadPool<T>();
    }
    return tp;
}

2、线程安全问题

上面的线程池的单例模式，看起来没有什么问题。可是当我们有多个线程去调用 getThreadPool函数，去创建线程池的时候，可能会有多个线程同时进入判断，判断出线程池指针为空，然后创建线程池对象。这样就会创建出多个线程池对象，这就不符合我们单例模式的要求了，所以我们必须让在同一时刻只有一个线程能够进入判断，我们就要用到锁了。

定义一个静态锁，并初始化：

private:
    vector<Thread *> threads_;
    int num_;
    queue<T> task_queue_;
    pthread_mutex_t lock;
    pthread_cond_t cond;
    static ThreadPool<T> *tp;
    static pthread_mutex_t lock;

// 类外初始化
​template <class T>
pthread_mutex_t ThreadPool<T>::lock = PTHREAD_MUTEX_INITIALIZER;

对 getThreadPool函数进行加锁：

    static ThreadPool<T> *getThreadPool()
    {
        if (tp == nullptr)
        {
            pthread_mutex_lock(&lock);
            if (tp == nullptr)
            {
                tp = new ThreadPool<T>();
            }
            pthread_mutex_unlock(&lock);
        }
        return tp;
    }

对于上面的代码：我们为什么要在获取锁之前还要再加一个判断指针为空的条件呢？

当已经有一个线程创建出来了线程池的单例模式后，在这之后的所有其他线程即使申请到锁，紧着着下一步就是去释放锁，它不会进入第二个 if 条件里面。其实这样是效率低下的，因为线程会频繁申请锁，然后就释放锁。所以我们在最外层再加一个if判断，就可以阻止后来的线程不用去申请锁创建线程池了，直接返回已经创建出来的线程池。

三、其他锁

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

2、乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
~ CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。

3、自旋锁：说到自旋锁，我们不得不说一说我们之前所用到的锁，我们之前所用的锁都是互斥锁，当线程没有竞争到互斥锁时，它会阻塞等待，只有等锁被释放了后，才能去重新申请锁。而对于自旋锁，当线程没有竞争到自旋锁的时候，线程会不断地循环检测去申请自旋锁，直到拿到锁。

一般来说，如果临界区的代码执行时间比较长的话，我们是使用互斥锁而不是自旋锁的，这样线程不会因为频繁地检测去申请锁而占用CPU资源。如果临界区的代码执行时间较短的话，我们一般就最好使用自旋锁，而不是互斥锁，因为互斥锁申请失败，是要阻塞等待，是需要发生上下文切换的，如果临界区执行的时间比较短，那可能上下文切换的时间会比临界区代码执行的时间还要长。