【Linux】线程池线程安全的单例模式和STL读者写者问题

需要云服务器等云产品来学习Linux的同学可以移步/–>腾讯云<–/官网，轻量型云服务器低至112元/年，优惠多多。（联系我有折扣哦）

文章目录

1. 线程池
- 1.1 线程池是什么
- 1.2 为什么要有线程池
- 1.3 线程池的应用场景
- 1.4 线程池的任务
- 1.5 线程池的代码实现
2. 线程安全的单例模式
3. STL、智能指针和线程安全
4. 其他常见锁的了解
5. 读者写者问题

1. 线程池

1.1 线程池是什么

线程池是一种线程的使用方式，是一种池化技术，在很早之前，我们其实接触过一些池化技术的例子，比如STL容器内部实现的alloctor就是一种内存的池化技术，将要用的空间首先申请一大批，然后再按照容器的需要从这个内存池中申请。还有在学习进程的时候，我们也尝试写过一些进程池的代码，一次创建一大批进程，按照要求分别执行不同的任务。当然这里也可以创建线程池，用线程来执行任务

1.2 为什么要有线程池

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

1.3 线程池的应用场景

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

1.4 线程池的任务

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

1.5 线程池的代码实现

/* ThreadPool.hpp 线程池的代码实现 */
#pragma once
#include "LockGuard.hpp"
#include "Thread.hpp"
#include <vector>
#include <queue>
#include <string>
#include <iostream>

const int gnum = 5; // 线程池中默认的线程个数

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;
};

/* 线程池类的实现 */
template <class T>
class ThreadPool
{
public:
    static void *handleTask(void *args) // 线程需要执行的回调函数
    {
        ThreadData<T> *td = static_cast<ThreadData<T> *>(args);
        while (true)
        {
            T t; // 构建任务对象
            {
                LockGuard lockGuard(td->threadpool->mutex()); // 上锁
                while (td->threadpool->isQueueEmpty())
                {
                    // 如果任务队列为空，线程挂起，等待队列中被填充任务
                    td->threadpool->threadWait();
                }
                t = td->threadpool->pop(); // 如果队列中有任务，就拿出任务
            }
            // 任务在锁外执行
            std::cout << "获取了一个任务" << t.toTaskString() << "并执行了任务，结果是 " << t() << std::endl;
        }
        delete td;
        return nullptr;
    }

public: // 给handleTask调用的外部接口
    pthread_mutex_t *mutex() { return &_mutex; }
    bool isQueueEmpty() { return _task_queue.empty(); }
    void threadWait() { pthread_cond_wait(&_cond, &_mutex); }
    T pop() // 获取线程池中任务队列里需要执行的下一个任务
    {
        T t = _task_queue.front();
        _task_queue.pop();
        return t;
    }

public:                               // 需要暴露给外部的接口
    ThreadPool(const int &num = gnum) // 构造函数，初始化互斥量和条件变量，构建指定个数的Thread对象
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
        for (int i = 0; i < num; ++i)
        {
            _threads.push_back(new Thread());
        }
    }
    void run() // 为所有线程对象创建真正的执行流，并执行对应的回调函数
    {
        for (const auto &thread : _threads)
        {
            ThreadData<T> *td = new ThreadData<T>(this, thread->GetTaskName()); // 构造handleTask的参数对象
            thread->start(handleTask, td);                                      // 调用该线程的start函数，创建新线程执行指定的handleTask任务
            std::cout << thread->GetTaskName() << " start..." << std::endl;
        }
    }
    void push(const T &in) // 将指定任务push到队列中
    {
        // 加锁
        LockGuard lockGuard(&_mutex); // 自动加锁，在当前代码段结束之后调用LockGuard的析构函数解锁
        _task_queue.push(in);
        pthread_cond_signal(&_cond); // 发送信号表示此时task_queue中有值，让消费者可以使用
    }
    ~ThreadPool() // 析构函数，销毁互斥量和条件变量，delete所有thread对象指针，自动调用thread对象的析构函数
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
        for (auto &thread : _threads)
        {
            delete thread;
        }
    }

private:
    std::vector<Thread *> _threads; // 保存所有线程对象的指针
    std::queue<T> _task_queue;      // 需要被分配的任务队列
    pthread_mutex_t _mutex;         // 任务队列需要被互斥的访问
    pthread_cond_t _cond;           // 生产任务和消费任务之间需要进行同步
};

构建和测试线程池用到的小组件

/* Task.hpp 线程对象，C++风格的线程库*/
#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <pthread.h>
#include <cassert>

class Thread
{
public:
    using func_t = std::function<void *(void *)>; // 定义func_t类型
    static int number;                            // 线程编号，按照一次运行时的调用次数计数
public:
    Thread()
    {
        char *buffer = new char[64];
        name_ = "thread-" + std::to_string(++number);
    }
    static void *start_routine(void *args)
    {
        Thread *_this = static_cast<Thread *>(args);
        void *ret = _this->run(_this->args_);
        return ret;
    }
    void *run(void *arg)
    {
        return func_(arg);
    }
    void start(func_t func, void *args)
    {
        func_ = func;
        args_ = args;
        int n = pthread_create(&tid_, nullptr, start_routine, this);
        assert(n == 0);
        (void)n;
    }
    void join()
    {
        int n = pthread_join(tid_, nullptr);
        assert(n == 0);
        (void)n;
    }
    std::string GetTaskName()
    {
        return name_;
    }
    ~Thread() {}

private:
    std::string name_; // 线程名
    pthread_t tid_;    // 线程id
    func_t func_;      // 线程调用的函数
    void *args_;       // 线程调用函数的参数
};
int Thread::number = 0;

/* LockGuard.hpp RAII风格的互斥锁 */
#pragma once

#include <pthread.h>

class Mutex
{
public:
    Mutex(pthread_mutex_t *lock_p = nullptr) : _lock_p(lock_p) {} // 构造函数

    void lock() // 加锁
    {
        if (_lock_p)
            pthread_mutex_lock(_lock_p);
    }
    void unlock() // 解锁
    {
        if (_lock_p)
            pthread_mutex_unlock(_lock_p);
    }

    ~Mutex() {} // 析构函数

private:
    pthread_mutex_t *_lock_p;
};

class LockGuard // RAII风格的锁的实现
{
public:
    LockGuard(pthread_mutex_t *mutex) : _mutex(mutex) // 构造函数
    {
        _mutex.lock(); // 在构造函数中加锁
    }
    ~LockGuard() // 析构函数
    {
        _mutex.unlock(); // 在析构函数中解锁
    }

private:
    Mutex _mutex;
};

/* Task.hpp 测试用到的测试任务 */
#pragma once

#include <string>
#include <iostream>
#include <functional>

static std::string oper = "+-*/%";

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), _callback(func)
    {
    }
    std::string operator()()
    {
        int result = _callback(_x, _y, _op);
        char buffer[64];
        snprintf(buffer, sizeof buffer, "%d %c %d = %d", _x, _op, _y, result);
        return buffer;
    }
    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 _callback;
};
class SaveTask
{
    typedef std::function<void(const std::string)> func_t;

public:
    SaveTask() {}
    SaveTask(const std::string &msg, func_t func)
        : _msg(msg), _func(func)
    {
    }
    void operator()()
    {
        _func(_msg);
    }

private:
    std::string _msg;
    func_t _func;
};
int mymath(int a, int b, char op)
{
    int ans = 0;
    switch (op)
    {
    case '+':
        ans = a + b;
        break;
    case '-':
        ans = a - b;
        break;
    case '*':
        ans = a * b;
        break;
    case '/':
    {
        if (b == 0)
        {
            std::cerr << "div zero error!" << std::endl;
            ans = -1;
        }
        else
            ans = a / b;
    }
    break;
    case '%':
    {
        if (b == 0)
        {
            std::cerr << "mod zero error!" << std::endl;
            ans = -1;
        }
        else
            ans = a % b;
    }
    break;
    default:
        break;
    }
    return ans;
}
void Save(const std::string &msg)
{
    FILE *fp = fopen("./log.txt", "a+");
    if (fp == NULL)
    {
        std::cerr << "open file error" << std::endl;
        return;
    }
    fputs(msg.c_str(), fp);
    fputs("\n", fp);

    fclose(fp);
}

测试代码：

/* main.cc */
#include "ThreadPool.hpp"
#include "Task.hpp"
#include <string>
#include <memory>
#include <unistd.h>

int main()
{
    // 创建线程用于等待输入，执行任务
    std::unique_ptr<ThreadPool<CalTask>> tp(new ThreadPool<CalTask>());
    tp->run();

    // 主线程用于输入任务
    int x, y;
    char op;
    while (true)
    {
        std::cout << "请输入数据1# ";
        std::cin >> x;
        std::cout << "请输入数据2# ";
        std::cin >> y;
        std::cout << "请输入你要进行的运算#";
        std::cin >> op;
        CalTask t(x, y, op, mymath);
        tp->push(t);
        sleep(1);
    }
    return 0;
}

运行结果：

2. 线程安全的单例模式

关于设计模式和单例模式的讲解，我们在C++专栏中已经有过了解，感兴趣的可以自行探究，这里附上链接【C++】特殊类设计

我们知道单例模式的设计有饿汉模式和懒汉模式两种，饿汉模式的实现会拖慢启动时间，所以这里我们采用懒汉模式来实现我们刚刚创建的线程池。

在一个进程中，有一个线程池即可，所以这里我们的线程池要改写成单例模式的，我们使用懒汉模式来改写

/* 懒汉模式的实现 */
// 头文件 ...

// ...

/* 线程池类的实现 */
template <class T>
class ThreadPool
{
public:
	// ...
public: // 给handleTask调用的外部接口
	// ...
public:                               // 需要暴露给外部的接口
	// ...
    static ThreadPool<T> *getInstance()
    {
        if(nullptr == tp)
        {
            std::lock_guard<std::mutex> lck(_singletonlock);
            if(nullptr == tp)
            {
                tp = new ThreadPool<T> ();
            }
        }
        return tp;
    }
private: // 单例模式需要私有化的接口
    ThreadPool(const int &num = gnum) // 构造函数，初始化互斥量和条件变量，构建指定个数的Thread对象
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
        for (int i = 0; i < num; ++i)
        {
            _threads.push_back(new Thread());
        }
    }
    //delete拷贝构造和析构函数
    ThreadPool(const ThreadPool<T> &) = delete;
    ThreadPool<T> *operator=(const ThreadPool<T> &) = delete;

private:
    std::vector<Thread *> _threads; // 保存所有线程对象的指针
    std::queue<T> _task_queue;      // 需要被分配的任务队列
    pthread_mutex_t _mutex;         // 任务队列需要被互斥的访问
    pthread_cond_t _cond;           // 生产任务和消费任务之间需要进行同步

    static ThreadPool<T> *tp; // 静态成员，存放ThreadPool指针
    static std::mutex _singletonlock; // 创建线程安全的单例对象要加的锁
};
template<class T>
ThreadPool<T> *ThreadPool<T>::tp = nullptr;
template<class T>
std::mutex ThreadPool<T>::_singletonlock;

3. STL、智能指针和线程安全

STL中的容器是否是线程安全的?

不是。原因是, STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响。而且对于不同的容器, 加锁方式的不同, 性能可能也不同(例如hash表的锁表和锁桶)。因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全

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

对于 unique_ptr, 由于只是在当前代码块范围内生效, 因此不涉及线程安全问题
对于 shared_ptr, 多个对象需要共用一个引用计数变量, 所以会存在线程安全问题. 但是标准库实现的时候考虑到了这个问题, 基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数

4. 其他常见锁的了解

**悲观锁：**在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。
自旋锁，公平锁，非公平锁？

5. 读者写者问题

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

读写锁的分析：读写锁一共是两个锁，分别为读锁和写锁，对应着读者和写者。当写者需要写数据时，请求写锁，然后再写数据，此时读者不能读，当写锁不存在时，多个读者可以并发的访问同一数据，提高了效率

读者写者问题和生产者消费者模型的本质区别就是消费者会取走数据，而读者不会取走数据。

//初始化读写锁
pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);

//销毁读写锁
pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

//读加锁
pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);

//写加锁
pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

//解锁
pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

本章完…

分别为读锁和写锁，对应着读者和写者。当写者需要写数据时，请求写锁，然后再写数据，此时读者不能读，当写锁不存在时，多个读者可以并发的访问同一数据，提高了效率

读者写者问题和生产者消费者模型的本质区别就是消费者会取走数据，而读者不会取走数据。

//初始化读写锁
pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);

//销毁读写锁
pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

//读加锁
pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);

//写加锁
pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

//解锁
pthread_rwlock_unlock(pthread_rwlock_t *rwlock);