线程池|单例模式|STL、智能指针线程安全|读者写者问题

线程池

线程池的逻辑思想：

每当我们处理一个任务就要创建一个线程，创建线程的开销是很大的。因此我们可以预先创建一批线程，任务队列里没有任务的时候，每个线程都休眠，当队里中有任务的时候，就可以唤醒线程进行处理。唤醒线程的成本比创建整个线程的成本小，这就是线程池的逻辑思想。

线程池的概念：

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

线程池常见的应用场景：

需要大量的线程来完成任务，且完成任务的时间比较短。

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

接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。

代码实现

首先我们先熟悉一下整体的框架：

Thread.hpp

#pragma once
#include <pthread.h>
#include <iostream>
#include <cassert>
#include <string>
#include <functional>
namespace ThreadNs
{
    typedef std::function<void*(void*)> func_t;
    const int num  =1024;
    class Thread
    {
    private:
        static void* start_routine(void*args)
        {
            Thread* _this = static_cast<Thread*>(args);
            return _this->callback();
        }
    public:
        Thread()
        {
            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;
        }
 
        void join()
        {
            int n = pthread_join(tid_,nullptr);
            assert(n==0);
            (void)n;
        }
 
        std::string threadname()
        {
            return name_;
        }
 
        ~Thread()
        {}
 
        void* callback()
        {
            return func_(args_);
        }
    private:
        std::string name_;
        func_t func_;
        void *args_;
        pthread_t tid_;
        static int threadnum;
    };
    int Thread::threadnum = 1;
}

 LockGuard.hpp

#include <iostream>
#include <mutex>
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
{
public:
    LockGuard(pthread_mutex_t*mutex)
    :mutex_(mutex)
    {
        mutex_.lock();
    }
 
    ~LockGuard()
    {
        mutex_.unlock();
    }
private:
    Mutex mutex_;
};

ThreadPool.hpp

#pragma once
#include "Thread.hpp"
#include "LockGuard.hpp"
#include <vector>
#include <queue>
#include <pthread.h>
#include <mutex>
#include <unistd.h>
using namespace ThreadNs;
const int gnum = 3;
template <class T>
class ThreadPool;
 
template <class T>
class ThreadData
{
public:
    ThreadPool<T> *threadpool;
    std::string name;
public:
    ThreadData(ThreadPool<T> *tp, const std::string &n) : threadpool(tp), name(n)
    { }
};
template <class T>
class ThreadPool
{
private:
    static void *handlerTask(void *args)
    {
        ThreadData<T> *td = (ThreadData<T> *)args;
        ThreadPool<T> *threadpool = static_cast<ThreadPool<T> *>(args);
        while (true)
        {
            T t;
            {
                LockGuard lockguard(td->threadpool->mutex());
                while(td->threadpool->isQueueEmpty())
                {
                    td->threadpool->threadWait();
                }
                t = td->threadpool->pop(); 
            }
            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 isQueueEmpty() { return _task_queue.empty(); }
    void threadWait() { pthread_cond_wait(&_cond, &_mutex); }
    T pop()
    {
        T t = _task_queue.front();
        _task_queue.pop();
        return t;
    }
    
    void Push(const T &in)
    {
        LockGuard lockguard(&_mutex);
        _task_queue.push(in);
        pthread_cond_signal(&_cond);
    }
 
    pthread_mutex_t *mutex()
    {
        return &_mutex;
    }
 
public:
    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());
        }
    }
 
    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;
        }
    }
 
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
        for (const auto &t : _threads)
            delete t;
    }
private:
    int _num;
    std::vector<Thread *> _threads;
    std::queue<T> _task_queue;
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;
};

我们将线程池进行了模板化，因此线程池当中存储的任务类型可以是任意的；现在我们想像之前处理各种数据的计算，那么先引入任务组件Task.hpp：

#pragma once
#include <iostream>
#include <functional>
class Task
{
    using func_t = std::function<int(int,int ,char)>;
public:
    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;
}

main.cc

#include "ThreadPool.hpp"
#include "Thread.hpp"
#include "Task.hpp"
#include <unistd.h>
#include <ctime>
int main()
{
   ThreadPool<Task>* tp = new ThreadPool<Task>();
   tp->run();
   srand(time(0));
   int x,y;
   char op;
   while(true)
   {
    x = rand()%10+1;
    y = rand()%20+1;
    op  =oper[rand()%oper.size()];
    Task t(x,y,op,mymath);
    tp->Push(t);
    sleep(1);
   }
    return 0;
}

单例模式

单例模式是一种创建型设计模式，它保证一个类只有一个实例存在，并且提供一个全局访问点来访问该实例。单例模式有饿汉模式和懒汉模式。

洗完的例子：

吃完饭 , 立刻洗碗 , 这种就是 饿汉方式 . 因为下一顿吃的时候可以立刻拿着碗就能吃饭 .

吃完饭 , 先把碗放下 , 然后下一顿饭用到这个碗了再洗碗 , 就是 懒汉方式 .

饿汉实现方式和懒汉实现方式

饿汉方式实现单例模式

template <typename T>
class Singleton {
static T data;
public:
static T* GetInstance() {
return &data;
}
};

懒汉方式实现单例模式

template <typename T>
class Singleton {
static T* inst;
public:
static T* GetInstance() {
if (inst == NULL) {
inst = new T();
}
return inst;
}
};

这样的懒汉设计可能会导致线程不安全，第一次调用 GetInstance 的时候, 如果两个线程同时调用, 可能会创建出两份 T 对象的实例。因此以下才是安全版的懒汉单例模式：

// 懒汉模式 , 线程安全

template <typename T>

class Singleton {

         volatile static T* inst; // 需要设置 volatile 关键字 , 否则可能被编译器优化 .

         static std::mutex lock;

public:

         static T* GetInstance()

    {

                  if (inst == NULL) { // 双重判定空指针 , 降低锁冲突的概率 , 提高性能 .

                         lock.lock(); // 使用互斥锁 , 保证多线程情况下也只调用一次 new.

                  if (inst == NULL) {

                         inst = new T();

                   }

            lock.unlock();

     }

                return inst;

    }

};

 线程池的单例模式

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

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

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

主函数的调用：

不过也许会出现多个线程同时申请资源的场景，所以还需要一把锁来保护这块资源，而这把锁也得设置成静态，因为GetSingle()函数是静态的：

 STL,智能指针和线程安全

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

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

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

 其他常见的各种锁

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

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

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

自旋锁：使用自旋锁时，当多线程发生竞争锁的情况时，加锁失败的线程会忙等待（这里的忙等待可以用 while 循环等待实现），直到它拿到锁。而互斥锁加锁失败后，线程会让出 CPU 资源给其他线程使用，然后该线程会被阻塞挂起。如果成功申请临界资源的线程，临界区代码执行时间过长，自旋的线程会长时间占用 CPU 资源，所以自旋的时间和临界区代码执行的时间是成正比的关系。如果临界区代码执行的时间很短，就不应该使用互斥锁，而应该选用自旋锁。因为互斥锁加锁失败，是需要发生上下文切换的，如果临界区执行的时间比较短，那可能上下文切换的时间会比临界区代码执行的时间还要长。

 读者和写者问题：

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

读写锁接口

//初始化读写锁
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);