Linux---多线程(下)

前情提要：Linux---多线程(上)

七、互斥

临界资源：多线程执行流共享的资源就叫做临界资源
临界区：每个线程内部，访问临界资源的代码，就叫做临界区
互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用
原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成

为什么要有互斥？什么情况下需要互斥？情景如下

加锁 (互斥锁)

接口如下：

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;// 锁被定义并初始化了
int ticket = 1000;

// 加锁
//1、尽可能的给少的代码块加锁，因为加锁本质是让线程线性执行该代码块，降低了运行效率
//2、一般加锁，都是给临界区加锁
void GetTicket(std::string name)
{
    while(1)
    {
        pthread_mutex_lock(&mutex);
        if(ticket > 0)
        {
            usleep(1000);
            printf("%s get a ticket : %d\n",name.c_str(),ticket);
            ticket--;
            pthread_mutex_unlock(&mutex);
        }
        else
        {
            pthread_mutex_unlock(&mutex);//确保锁要被释放
            break;
        }
    }
}

申请锁本身是安全的，原子的，因为锁也是公共资源，而我们是为了维护公共资源才创建的锁，如果申请锁的操作不是原子的，就会出问题
临界资源的访问要加锁，是由程序员保证的！！！
根据互斥的定义，任何时刻，只允许一个线程申请成功，多个线程申请失败，失败的线程在mutex上进行阻塞，本质就是等待
一个线程在临界区中访问临界资源的时候，可不可能发生切换？可能，加锁只是保证在一个线程执行临界区代码时，其他线程不能执行临界区代码，并不意味了其他线程的其他代码不能执行(就比如break语句其他线程就能执行)，所以可以切换，所以在一定程度上说，临界区代码的执行也是原子的

当然除了定义全局的锁，也可以定义局部的锁，如下(全局的锁也可以用下面的函数初始化和销毁，但建议用宏完成---不用手动销毁)

int main()
{
    pthread_mutex_t mutex;
    pthread_mutex_init(&mutex, nullptr);
    //哪个线程需要，就传给哪个线程
    //...
    pthread_mutex_destroy(&mutex);
    return 0;
}

但是加锁后，个别系统会出现很多票被同一个线程抢完的情况(如下图)，如果有线程长时间得不到资源，就会造成饥饿问题，解决饥饿问题，需要让线程执行具有一定的顺序性，即同步

当然我们可以封装一下锁，让它能被申请完之后，能自动释放

class LockGuard
{
public:
    LockGuard(pthread_mutex_t* m)
    :pmutex(m)
    {
        pthread_mutex_lock(pmutex);
    }
    
    ~LockGuard()
    {
        pthread_mutex_unlock(pmutex);
    }
private:
    pthread_mutex_t* pmutex;
};

加锁的原理

为了实现互斥锁操作，大多数体系结构都提供了swap或exchange指令，该指令的作用是把寄存器和内存单元的数据相交换，由于只有一条指令，保证了原子性，即使是多处理器平台，访问内存的总线周期也有先后，一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。(就拿一个处理器来考虑)

使用锁的原则：谁加锁，谁解锁

八、线程安全vs可重入(了解基本概念即可)

线程安全：多个线程并发同一段代码时，不会出现不同的结果。常见对全局变量或者静态变量进行操作，并且没有锁保护的情况下，会出现该问题。
重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数。

常见的线程不安全的情况

不保护共享变量的函数
函数状态随着被调用，状态发生变化的函数
返回指向静态变量指针的函数
调用线程不安全函数的函数

常见的线程安全的情况

每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的
类或者接口对于线程来说都是原子操作
多个线程之间的切换不会导致该接口的执行结果存在二义性

常见不可重入的情况

调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
可重入函数体内使用了静态的数据结构

常见可重入的情况

不使用全局变量或静态变量
不使用用malloc或者new开辟出的空间
不调用不可重入函数
不返回静态或全局数据，所有数据都有函数的调用者提供
使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据

可重入与线程安全联系

函数是可重入的，那就是线程安全的
函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的。

可重入与线程安全区别

可重入函数是线程安全函数的一种
线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生死锁，因此是不可重入的

九、死锁

概念介绍

死锁是指在一组进程中的各个进程(或者线程)均占有不会释放的资源，但因互相申请被其他进程(或者线程)所占用不会释放的资源而处于的一种永久等待状态，如下

那么一个线程申请资源可不可能出现死锁的情况呢？？？当然可能，如果我们在释放锁的时候，将代码写成了申请锁，那么线程就会自己把自己阻塞住，从而形成死锁

死锁的四个必要条件

互斥条件：一个资源每次只能被一个执行流使用
请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
不剥夺条件：一个执行流已获得的资源，在末使用完之前，不能强行剥夺
循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系

(可以带入上面那个图进行验证)

如何避免死锁

破坏死锁的四个必要条件

破坏条件1：不用锁
破坏条件2：如果申请不到锁，就释放自己申请到的锁
破坏条件3：强制让其他线程/进程释放锁
破坏条件4：保持申请锁的顺序一致

避免锁未释放的场景

资源一次性分配

避免死锁的算法：死锁检测算法(了解) 银行家算法（了解）【这里不做介绍】

十、条件变量

概念介绍

同步：在保证数据安全的前提下，让执行流能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步
条件变量是保持同步的一种方式，在资源就绪之前，它让申请资源的线程阻塞等待，不要一直加锁解锁访问资源，一旦资源就绪，就会满足条件，它会唤醒线程去访问资源，比如上下课有铃声提醒，不用我们一直去看时间，一旦铃声响了，我们就知道要上课/下课了。

条件变量可以理解为下面这样一个结构体
struct cond
{
    int flag;//是否满足条件
    struct tcb* wait_q;//等待队列
}

十一、生产消费模型

概念介绍

基于BlockingQueue的生产者消费者模型

//BlockQueue.hpp
#include <iostream>
#include <queue>
#include <pthread.h>

const int N = 5;
template<class T>
class BlockQueue
{
public:
    BlockQueue(int capacity = N)
        :_capacity(capacity)
    {
        pthread_mutex_init(&_mutex,nullptr);
        pthread_cond_init(&_c_cond,nullptr);
        pthread_cond_init(&_p_cond,nullptr);
    }
    
    bool IsFull()
    {
        return _capacity == _bq.size();
    }

    bool IsEmpty()
    {
        return _bq.empty();
    }

    void push(const T& in) //生产者
    {
        pthread_mutex_lock(&_mutex);
        // if(IsFull()) // 可能会出现问题：如果多个线程同时被唤醒(pthread_cond_broadcast) ，但是只有一份资源，就会出现队列为空pop
        // 如果理解不了，可以认为等待函数可能调用失败
        while(IsFull()) // 防止出现"伪苏醒"情况,即上面两种情况
        {
            //阻塞等待 消费者消费
            pthread_cond_wait(&_p_cond,&_mutex);
        }

        _bq.push(in);
        pthread_cond_signal(&_c_cond); // 唤醒消费线程
        pthread_mutex_unlock(&_mutex);
    }

    void pop(T* out) //消费者
    {
        pthread_mutex_lock(&_mutex);
        while(IsEmpty())
        {
            //阻塞等待 生产者生产
            pthread_cond_wait(&_c_cond,&_mutex);
        }

        *out = _bq.front(); _bq.pop();
        pthread_cond_signal(&_p_cond); // 唤醒生产线程
        pthread_mutex_unlock(&_mutex);
    }

   ~BlockQueue()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_c_cond);
        pthread_cond_destroy(&_p_cond);
    }

private:
    std::queue<T> _bq;
    int _capacity;
    pthread_mutex_t _mutex;
    pthread_cond_t _c_cond;
    pthread_cond_t _p_cond;

    // 可以设置相应的唤醒策略
    // int consumer_water_line;
    // int productor_water_line;
};

//Task.hpp
#include <iostream>
#include <string>
enum
{
    ok,
    zero,
    unknow
};

class Task
{
public:
    Task() = default;
    Task(int x, int y, char op)
        : _data_x(x), _data_y(y), _op(op)
    {}

    void Run()
    {
        switch (_op)
        {
        case '+':
            _result = _data_x + _data_y;
            break;
        case '-':
            _result = _data_x - _data_y;
            break;
        case '*':
            _result = _data_x * _data_y;
            break;
        case '/':
            if(_data_y==0) _code = zero;
            else _result = _data_x / _data_y;
            break;
        case '%':
            if(_data_y==0) _code = zero;
            else _result = _data_x % _data_y;
            break;
        default:
            _code = unknow;
            break;
        }
    }

    void operator()()
    {
        Run();
    }

    std::string PrintResult()
    {
       return std::to_string(_data_x) + _op +std::to_string(_data_y) + " = " + std::to_string(_result) + " [" + std::to_string(_code) +"]";
    }

    std::string PrintTask()
    {
       return std::to_string(_data_x) + _op +std::to_string(_data_y) + " = ?";
    }
private:
    int _data_x;
    int _data_y;
    char _op;
    int _result;
    int _code = ok;
};


//test.cc
#include "BlockQueue.hpp"
#include "Task.hpp"
#include <time.h>
#include <unistd.h>
const std::string ops = "+-*/()&^|";

void* Consumer(void* args) //生产线程
{
    BlockQueue<Task>* bq = static_cast<BlockQueue<Task>*>(args);
    while(1)
    {
        sleep(1);
        Task t(rand()%10,rand()%10,ops[rand()%ops.size()]); // [1,10]
        bq->push(t);
        std::cout << "Task : " << t.PrintTask() << std::endl;
    }
    return nullptr;
}

void* Productor(void* args) //消费线程
{
    BlockQueue<Task>* bq = static_cast<BlockQueue<Task>*>(args);
    while(1)
    {
        Task t;
        bq->pop(&t);
        t();
        std::cout << "result : " << t.PrintResult() << std::endl;
    }
    return nullptr;
}

int main()
{
    srand(time(nullptr));
    BlockQueue<Task>* bq = new BlockQueue<Task>();
    pthread_t t1,t2;
    pthread_create(&t1,nullptr,Consumer,(void*)bq);
    pthread_create(&t2,nullptr,Productor,(void*)bq);

    pthread_join(t1,nullptr);
    pthread_join(t2,nullptr);
    return 0;
}

对生产消费模型的进一步理解：

1、在超市(内存空间，在上面代码中是队列)中交换的可以是基本数据，也可以是类对象

2、如何理解生产消费模型是高效的？根据上面的代码，我们只能看出生产和消费在相互牵制，一旦资源满了，只能等消费者消费，一旦资源处理完了，只能等生产者生产，似乎效率并没有变高。

但是数据是从哪里来的呢？数据又是如何处理的呢？这里说的高效主要体现在生产者在产生数据，消费者在处理数据时是独立的，可以并发/并行的(上面的例子由于数据处理比较简单，看不出效果)

上面代码是单线程生产，单线程消费，如何将它改为多线程生产，多线程消费呢？？？

在单-单生产消费模型中，我们只要考虑生产者和消费者之间的互斥同步关系即可，但如果是多-多生产消费模型，我们就需要多考虑生产者之间和消费者之间的互斥关系了。但是我们上面的代码中只用了一个锁，也就是说每个线程在访问资源时都是互斥关系，符合条件，所以我们写的BlockQueue也能支持多线程的生产消费模型，大家可以多创建几个生产线程和消费线程验证一下

十二、POSIX信号量

概念介绍

与进程通讯中提到的System V信号量的作用是一样的，都是用于同步操作，它们的用法有区别，有兴趣可以去了解一下，这里仅仅介绍POSIX信号量的用法

在进程间通信(下)中，我们介绍过信号量：

信号量的本质是一个计数器
申请信号量本质就是预定资源
PV操作是原子的(简单理解P就是对计数器做--操作，V就是对计数器做++操作)

信号量和锁的区别(以BlockQueue为例)：锁是将资源当作整体来使用，我们访问BlockQueue的时候，只允许该队列一次被一个线程访问，但实际上，一个线程每次只需要该队列中的一个资源即可，没必要将整个队列都锁住，而信号量是将资源分成一个个局部来使用，即该队列一次可以被多个线程访问，只要访问的资源不重复即可，同时，一旦申请信号量成功就意味着该线程一定能拿到资源，不需要再去判断是否有资源。

基于环形队列的生产消费模型

(环形队列这里就不做介绍了，可以用数组或者链表实现)

代码如下

// RingQueue.hpp
#include <iostream>
#include <pthread.h>
#include <semaphore.h>
#include <vector>
#include "LockGurad.hpp"
const int N = 5;
template <class T>
class RingQueue
{
public:
    RingQueue(int n = N)
        : _rq(N), _p_idx(0), _c_idx(0)
    {
        pthread_mutex_init(&_p_mutex,nullptr);
        pthread_mutex_init(&_c_mutex,nullptr);
        sem_init(&_space_sem, 0, n);
        sem_init(&_data_sem, 0, 0);
    }

    void P(sem_t &sem)
    {
        sem_wait(&sem);
    }

    void V(sem_t &sem)
    {
        sem_post(&sem);
    }

    void push(const T &in)
    {
        // 先申请信号量，在申请锁，可以先放部分线程进来竞争锁，这样能减少锁的锁的竞争，并且不用判断资源是否足够

        // 先申请锁，在申请信号量，线程需要竞争完锁，在去申请信号量，可以是可以，但是却失去了信号量的功能，因为申请信号量时只会有一个线程，其他线程都在等待锁，无法申请信号量，和定义一把锁是一样的效果

        // 可以理解为参观博物馆，我们先去预约买票，在进去参观，和 在要进去参观时才开始买票  两种模式

        P(_space_sem);
        {
            LockGuard lock(&_p_mutex);// 要访问临界资源，保证生产者间的互斥关系，如果是单线程生产可以不加锁
            _rq[_p_idx++] = in;
            _p_idx %= _rq.size();
        }
        V(_data_sem);
    }

    void pop(T *out)
    {
        P(_data_sem);
        {
            LockGuard lock(&_c_mutex);// 要访问临界资源，保证消费者间的互斥关系，如果是单线程消费可以不加锁
            *out = _rq[_c_idx++];
            _c_idx %= _rq.size();
        }
        V(_space_sem);
    }

    ~RingQueue()
    {
        pthread_mutex_destroy(&_p_mutex);
        pthread_mutex_destroy(&_c_mutex);
        sem_destroy(&_space_sem);
        sem_destroy(&_data_sem);
    }

private:
    std::vector<T> _rq;
    pthread_mutex_t _p_mutex; //保持生产者间的互斥
    pthread_mutex_t _c_mutex; //保持消费者间的互斥
    sem_t _space_sem;
    sem_t _data_sem;
    int _p_idx;
    int _c_idx;
};

十三、线程池

线程池

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

线程池的应用场景：

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

线程池的种类（线程池示例）：

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

// ThreadPool.hpp

#pragma once
#include <iostream>
#include <queue>
#include <vector>
#include <pthread.h>
#include "thread.hpp"
#define N 5

struct ThreadDate
{
    ThreadDate(const std::string& name):_threadname(name)
    {}
    ~ThreadDate()
    {}

    std::string _threadname;
};

template<class T>
class ThreadPool
{
private:
    ThreadPool(const ThreadPool&tmp) = delete;
    ThreadPool& operator=(const ThreadPool&tmp) = delete;

    ThreadPool(int n = N):thread_num(n)
    {
        pthread_mutex_init(&_mutex,nullptr);
        pthread_cond_init(&_cond,nullptr);
        
        for(int i=0;i<thread_num;i++)
        {
            std::string name = "thread-" + std::to_string(i);
            ThreadDate td(name);
            _threads.emplace_back(name,std::bind(&ThreadPool::ThreadRun,this,std::placeholders::_1),td);
        }
    }

public:
    //单例模式
    static ThreadPool* GetInstance()
    {
        if(_instance == nullptr)
        {
            LockGuard lock(&mtx);
            if(_instance==nullptr)
            {
                _instance = new ThreadPool<T>();
            }
        }
        return _instance;
    }


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

    void ThreadRun(ThreadDate& td)
    {
        // while(1)
        // std::cout<<"thread is running "<<std::endl;
        while(1)
        {
            T t;// 这里的任务用的是上面的Task
            {
                LockGuard lock(&_mutex);
                while(_q.empty())
                    pthread_cond_wait(&_cond,&_mutex);
                t = _q.front(); // 获取任务
                _q.pop();
            }
            t(); // 任务类实现的处理任务的仿函数，具体结合自己的任务进行函数调用
            lg(Info,"%s is running, result is %s",td._threadname.c_str(),t.PrintResult().c_str());// 打印日志，任务类提供的函数接口，具体结合自己的任务进行函数调用
        }
    }

    void Start()
    {
        for(auto &thd: _threads)
        {
            thd.Start();
        }
    }

    void Wait()
    {
        for(auto &thd: _threads)
        {
            thd.Join();
        }
    }


    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }

private:
    std::queue<T> _q; // 任务队列
    std::vector<thread<ThreadDate>> _threads; // 线程池
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;
    int thread_num; // 线程个数

    static ThreadPool<T>* _instance;
    static pthread_mutex_t mtx;
};

template<class T>
ThreadPool<T>* ThreadPool<T>::_instance = nullptr;

template<class T>
pthread_mutex_t ThreadPool<T>::mtx = PTHREAD_MUTEX_INITIALIZER;

日志

1、可以向显示器打印，也可以向文件中写入

2、包含：时间、内容、日志等级、文件名等等相关数据

// Log.hpp
#pragma once
#include <iostream>
#include <cstdio>
#include <cstring>
#include <cstdarg>
#include <fstream>
#include <ios>
#include <sys/stat.h>
#include <pthread.h>
#include <unistd.h>
#include "LockGuard.hpp"

enum
{
    Debug,
    Info,
    Warning,
    Error,
    Fatal
};

enum
{
    Screem = 10,
    OneFile,
    Files
};

const int defaultstyle = Screem;
const std::string filename = "Log";
const std::string dir = "Log";

std::string LevelToString(int level)
{
    switch (level)
    {
    case Debug:
        return "Debug";
    case Info:
        return "Info";
    case Warning:
        return "Warning";
    case Error:
        return "Error";
    case Fatal:
        return "Fatal";
    default:
        return "unknown";
    }
}

class Log
{
public:
    Log():_style(defaultstyle),_filename(filename),_filepath(dir)
    {
        mkdir(_filepath.c_str(),0775); // 在当前目录下创建目录
        pthread_mutex_init(&_mutex,nullptr);
    }

    std::string local_time()
    {
        time_t cur = time(nullptr);
        struct tm* t = localtime(&cur);
        char buffer[128];
        // asctime_r(t, buffer);
        snprintf(buffer,sizeof(buffer),"%d-%d-%d %d:%d:%d",t->tm_year,t->tm_mon,t->tm_mday,t->tm_hour,t->tm_min,t->tm_sec);
        return buffer;
    }

    void Write(const std::string &info, const std::string &suffix)
    {
        // 可以加锁，保证线程安全 ...
        LockGuard lock(&_mutex);
        std::string name = _filepath + "/" + _filename + suffix;
        std::ofstream ifs(name.c_str(), std::ios_base::out | std::ios_base::app);
        if(ifs.is_open())
            ifs<<info;
        ifs.close();
    }

    void WriteToFile(const std::string &info, const std::string& level)
    {
        switch(_style)
        {
            case Screem:
                std::cout << info;
                break;
            case OneFile:
                Write(info,".all");
                break;
            case Files:
                Write(info,"."+level);
                break;
            default:
                break;
        }
    }

    void Message(int level, const char *format, ...)
    {
        char buffer[1024];
        va_list args;
        va_start(args, format);
        vsnprintf(buffer, sizeof(buffer), format, args);
        va_end(args);
        // printf("[%s][%s][%s]\n",LevelToString(level).c_str(),local_time().c_str(),buffer);
        char info[4096];
        std::string lev = LevelToString(level);
        snprintf(info,sizeof(info),"[%s][%s][%s] %s\n",std::to_string(getpid()).c_str(),
            lev.c_str(),local_time().c_str(),buffer);
        WriteToFile(info,lev);
    }

    // 将Message函数转换成仿函数，方便调用
    void _Message_(int level, const char *format, va_list args)
    {
        char buffer[1024];
        vsnprintf(buffer, sizeof(buffer), format, args);
        char info[4096];
        std::string lev = LevelToString(level);
        snprintf(info,sizeof(info),"[%s][%s][%s] %s\n",std::to_string(getpid()).c_str(),
            lev.c_str(),local_time().c_str(),buffer);
        WriteToFile(info,lev);
    }

    void operator()(int level, const char *format, ...)
    {
        va_list args;
        va_start(args, format);
        _Message_(level,format,args);
        va_end(args);
    }

    ~Log()
    {
        pthread_mutex_destroy(&_mutex);
    }

    // 提供接口，方便我们改变日志的输出
    void Enable(int mode)
    {
        _style = mode;
    }

private:
    int _style;
    const std::string _filename;
    const std::string _filepath;
    pthread_mutex_t _mutex;
};

Log lg;

//test.cpp---测试代码
#include "LockGuard.hpp"
#include "Log.hpp"
#include "ThreadPool.hpp"
#include "thread.hpp"
#include "Task.hpp"
#include <time.h>

const std::string opers = "+-*/";
int main()
{
    srand(time(nullptr));
    ThreadPool<Task>::GetInstance()->Start();
    while(1)
    {
        int x = rand()%200+1;
        int y = rand()%200+1;
        int i = rand()%opers.size();
        Task t(x,y,opers[i]);
        ThreadPool<Task>::GetInstance()->push(t);
        usleep(1000);
        lg(Info,"task : %s",t.PrintTask().c_str());
    }
    ThreadPool<Task>::GetInstance()->Wait();
    return 0;
}