【Linux系统】POSIX信号量线程池

PISIX信号量

概念

POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。但POSIX可以用于线程间同步。

引入环形队列的概念

环形队列：当队列为空||为满时

head == end，我们发现这样无法区分为空为满两种情况

所以：我们需要1.计数器 2.牺牲一个空位置，判满就变为if(head == end + 1)

如果队列不为空不为满就是head != end

问题：多线程如何在环形队列中进行生产和消费1.单生产，单消费 2.多生产，多消费

1.队列为空，让谁先访问？生产者先生产

2.队列为满，让谁先访问？消费者来消费

3.队列不为空&&队列不为满，生产和消费同时进行！！！（并发的！！！）

第三点要满足条件:a.不然生产者把消费者套一个圈b.不能让消费者超过生产者

所以前面两点不仅要有顺序，还要互斥。

三点放在一起就是互斥与同步的！！！ ---- 所以这些信号量就能做到这些！！！

消费者只关心的资源，数据资源！生产者只关心的资源，空间资源！！！

sem_t data_sem = 0;
sem_t space_sem = N;

生产与消费的关系是通过PV操作来实现的！

//生产者
P(space_sem);
//生产
V(data_sem);

//消费者
P(data_sem);
//消费
V(space_sem);

实现

信号量接口

初始化信号量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);

sem：指向信号量的指针
pshared:0表示线程间共享，非零表示进程间共享
value：信号量初始值

销毁信号量

int sem_destroy(sem_t *sem);

等待信号量（P操作）

功能：等待信号量，会将信号量的值减1

会将信号量的值减一，如果信号量的值为零，调用线程将被阻塞，直到信号量的值大于零

int sem_wait(sem_t *sem); //P()

发布信号量（V操作）

功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。

会将信号量的值加一，如果有任何线程在等待该信号量，它们将被唤醒。

int sem_post(sem_t *sem);//V()

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

生产消费模型的高效率核心问题是：构建数据和后来的处理数据实现了多线程并发执行！！！因为这两个部分是最耗时的！

环形队列采用数组模拟，用模运算来模拟环状特性
环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置，作为满的状态
但是我们现在有信号量这个计数器，就很简单的进行多线程间的同步过程

RingQueue.hpp

#pragma once
#include<iostream>
#include<vector>
#include<semaphore.h>
template<class T>
class RingQueue
{
    void P(sem_t& s)
    {
        sem_wait(&s);
    }
    void V(sem_t& s)
    {
        sem_post(&s);
    }
public:
    RingQueue(int max_cap):_max_cap(max_cap), _ring_queue(max_cap), _c_step(0), _p_step(0)
    {
        sem_init(&_data_sem, 0, 0);
        sem_init(&_space_sem, 0, max_cap);

        pthread_mutex_init(&_c_mutex, nullptr);
        pthread_mutex_init(&_p_mutex, nullptr);
    }
    void Pop(T* out)//消费
    {
        //信号量：信号量是一个计数器，是资源的预订机制。预订：在外部，可以不判断资源是否满足，就可以知道内部资源的情况。
        P(_data_sem);//信号量这里，对资源进行使用，申请，为什么不判断一下条件是否满足？？？信号量本身就是判断条件！！！
        pthread_mutex_lock(&_c_mutex);
        *out = _ring_queue[_c_step];
        _c_step++;
        _c_step %= _max_cap;
        pthread_mutex_unlock(&_c_mutex);
        V(_space_sem);
    }
    void Push(const T &in)//生产
    {
        P(_space_sem);
        pthread_mutex_lock(&_p_mutex);
        _ring_queue[_p_step] = in;
        _p_step++;
        _p_step %= _max_cap;
        pthread_mutex_unlock(&_p_mutex);
        V(_data_sem);
    }
    ~RingQueue()
    {
        sem_destroy(&_data_sem);
        sem_destroy(&_space_sem);

        pthread_mutex_destroy(&_c_mutex);
        pthread_mutex_destroy(&_p_mutex);
    }
private:
    std::vector<T> _ring_queue;
    int _max_cap;

    int _c_step;
    int _p_step;

    sem_t _data_sem;//消费者关心
    sem_t _space_sem;//生产者关心

    pthread_mutex_t _c_mutex;
    pthread_mutex_t _p_mutex;
};

线程池

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

下面我们使用打印日志的方式，并且运用单例模式中的懒汉模式实现了一个简单的线程池！！！

ThreadPool.hpp --- 线程池主逻辑封装

#pragma once
#include <iostream>
#include <queue>
#include <vector>
#include <string>
#include <unistd.h>
#include "Thread.hpp"
#include "Task.hpp"
#include "Log.hpp"

using namespace ThreadMoudle;
using namespace log_ns;
static const int gdefaultnum = 5;
void test()
{
    while (true)
    {
        std::cout << "hello world" << std::endl;
        sleep(1);
    }
}
template <typename T>
class ThreadPool
{
private:
    void LockQueue()
    {
        pthread_mutex_lock(&_mutex);
    }
    void UnlockQueue()
    {
        pthread_mutex_unlock(&_mutex);
    }
    void Wakeup()
    {
        pthread_cond_signal(&_cond);
    }
    void WakeupAll()
    {
        pthread_cond_broadcast(&_cond);
    }
    void Sleep()
    {
        pthread_cond_wait(&_cond, &_mutex);
    }
    bool IsEmpty()
    {
        return _task_queue.empty();
    }
    void HandlerTask(const std::string &name) // this
    {
        while (true)
        {
            // 取任务
            LockQueue();
            while (IsEmpty() && _isrunning)
            {
                _sleep_thread_num++;
                LOG(INFO, "%s thread sleep begin!\n", name.c_str());
                Sleep();
                LOG(INFO, "%s thread wakeup begin!\n", name.c_str());
                _sleep_thread_num--;
            }
            // 只需判断一种情况
            if (IsEmpty() && !_isrunning)
            {
                UnlockQueue();
                LOG(INFO, "%s thread quie!\n", name.c_str());
                break;
            }
            // 有任务
            T t = _task_queue.front();
            _task_queue.pop();
            UnlockQueue();
            // 处理任务,此处不用/不能在临界区中处理
            t();
            // std::cout << name << ":" << t.result() << std::endl;
            LOG(DEBUG, "hander task done, task is:%s\n", t.result().c_str());
        }
    }
    void Init()
    {
        func_t func = std::bind(&ThreadPool::HandlerTask, this, std::placeholders::_1);
        for (int i = 0; i < _thread_num; i++)
        {
            std::string threadname = "thread-" + std::to_string(i + 1);
            _threads.emplace_back(threadname, func);
            LOG(DEBUG, "construct thread %s done, init success!\n", threadname.c_str());
        }
    }
    void Start()
    {
        _isrunning = true;
        for (auto &thread : _threads)
        {
            LOG(DEBUG, "start thread %s done.\n", thread.Name().c_str());
            thread.Start();
        }
    }
    ThreadPool(int thread_num = gdefaultnum) : _thread_num(thread_num), _isrunning(false), _sleep_thread_num(0)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
    }
    ThreadPool(const ThreadPool<T> &) = delete;
    ThreadPool<T> operator=(const ThreadPool<T> &) = delete;

public:
    void Stop()
    {
        LockQueue();
        _isrunning = false;
        WakeupAll();
        UnlockQueue();
        LOG(DEBUG, "Thread Pool Stop Success!\n");
    }
    // 如果是多线程获取单例呢？
    static ThreadPool<T> *GetInstance()
    {
        LockGuard lockguard(&_sig_mutex);
        if (_tp == nullptr)
        {
            if (_tp == nullptr)
            {
                LOG(INFO, "create threadpool!\n");
                // thread-1,thread-2...
                _tp = new ThreadPool();
                _tp->Init();
                _tp->Start();
            }
            else
            {
                LOG(INFO, "get threadpool!\n");
            }
        }
        return _tp;
    }
    void Equeue(const T &in)
    {
        LockQueue();
        if (_isrunning)
        {
            _task_queue.push(in);
            if (_sleep_thread_num > 0)
                Wakeup();
        }
        UnlockQueue();
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }

private:
    int _thread_num;
    std::vector<Thread> _threads;
    std::queue<T> _task_queue;
    bool _isrunning;

    int _sleep_thread_num;

    pthread_mutex_t _mutex;
    pthread_cond_t _cond;

    // 单例模式
    static ThreadPool<T> *_tp;
    static pthread_mutex_t _sig_mutex;
};

template <class T>
ThreadPool<T> *ThreadPool<T>::_tp = nullptr;
template <class T>
pthread_mutex_t ThreadPool<T>::_sig_mutex = PTHREAD_MUTEX_INITIALIZER;

Thread.hpp --- 封装了我们自己的线程库

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

namespace ThreadMoudle
{
    //线程要执行的方法
    //typedef void (*func_t)(ThreadData *td);
    using func_t = std::function<void(const std::string&)>;
    class Thread
    {
    public:
        void Excute()
        {
            _isrunning = true;
            _func(_name);
            _isrunning = false;
        }
    public:
        Thread(const std::string& name, func_t func) :_name(name), _func(func)
        {
        }
        static void* ThreadRoutine(void* args)//新线程都会执行该方法
        {
            Thread* self = static_cast<Thread*>(args);//获得了当前对象
            self->Excute();
            return nullptr;
        }
        bool Start()
        {
            int n = ::pthread_create(&_tid, nullptr, ThreadRoutine, this);
            if(n != 0) return false;
            return true;
        }
        std::string Status()
        {
            if(_isrunning) return "running";
            else return "sleep";
        }
        void Stop()
        {
            if(_isrunning)
            {
                ::pthread_cancel(_tid);
                _isrunning = false;
            }
        }
        std::string Name()
        {
            return _name;
        }
        void Join()
        {
            ::pthread_join(_tid, nullptr);
        }
        ~Thread()
        {

        }
    public:
        std::string _name;
        pthread_t _tid;
        bool _isrunning;
        func_t _func; //线程要执行的回调函数
    };   
};

main.cc --- 不断推送任务到线程池并且测试

#include"ThreadPool.hpp"
#include"Task.hpp"
#include"Log.hpp"

using namespace log_ns;

int main()
{
    EnableScreen();
    // LOG(DEBUG, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // LOG(WARNING, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // LOG(FATAL, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // LOG(ERROR, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // EnableFile();
    // LOG(DEBUG, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // LOG(DEBUG, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // LOG(DEBUG, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);
    // LOG(DEBUG, "hello %d, world: %c, hello: %f\n", 1000, 'A', 3.14);

    // ThreadPool<Task>* tp = new ThreadPool<Task>();
    // tp->Init();
    // tp->Start();
    int cnt = 10;
    while(cnt)
    {
        //不断向线程池推送任务
        sleep(1);
        Task t(1, 1);
        ThreadPool<Task>::GetInstance()->Equeue(t);
        LOG(INFO, "equeue a task, %s\n", t.debug().c_str());
        sleep(1);
        cnt--;
    }
    ThreadPool<Task>::GetInstance()->Stop();
    LOG(INFO, "thread pool stop!\n");
    return 0;
}

Task.hpp --- 任务的来源我们这里是一个加法任务

#pragma once
#include<iostream>
#include<string>
#include<functional>

//typedef std::function<void()> task_t;
// using task_t = std::function<void()>;

// void Download()
// {
//     std::cout << "我是一个下载任务" << std::endl;
// }
class Task
{
public:
    Task()
    {}
    Task(int x, int y):_x(x), _y(y)
    {}
    void Excute()
    {
        _result = _x + _y;
    }
    std::string debug()
    {
        std::string msg = std::to_string(_x) + "+" + std::to_string(_y) + "=?";
        return msg;
    }
    void operator()()
    {
        Excute();
    }
    std::string result()
    {
        std::string msg = std::to_string(_x) + "+" + std::to_string(_y) + "=" + std::to_string(_result);
        return msg;
    }
private:
    int _x;
    int _y;
    int _result;
};

Log.hpp --- 封装一个我们自己的日志库

#pragma once
#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <cstdarg>
#include <fstream>
#include <cstring>
#include <pthread.h>
#include <cstring>
#include "LockGuard.hpp"

namespace log_ns
{
    enum
    {
        DEBUG = 1,
        INFO,
        WARNING,
        ERROR,
        FATAL // 致命的
    };
    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";
        }
    }
    std::string GetCurrTime()
    {
        time_t now = time(nullptr);
        struct tm *curr_time = localtime(&now);
        char buffer[128];
        snprintf(buffer, sizeof(buffer), "%04d-%02d-%02d %02d:%02d:%02d",
                 curr_time->tm_year + 1900,
                 curr_time->tm_mon + 1,
                 curr_time->tm_mday,
                 curr_time->tm_hour,
                 curr_time->tm_min,
                 curr_time->tm_sec);
        return buffer;
    }
    class logmessage
    {
    public:
        std::string _level;
        pid_t _id;
        std::string _filename;
        int _filenumber;
        std::string _curr_time;
        std::string _message_info;
    };

#define SCREEN_TYPE 1
#define FILE_TYPE 2

    const std::string glogfile = "./log.txt";
    pthread_mutex_t glock = PTHREAD_MUTEX_INITIALIZER;

    class log
    {
    public:
        log(const std::string &logfile = glogfile) : _logfile(logfile), _type(SCREEN_TYPE)
        {
        }
        void Enable(int type)
        {
            _type = type;
        }
        void FlushLogToScreen(const logmessage &lg)
        {
            printf("[%s][%d][%s][%d][%s] %s",
                   lg._level.c_str(),
                   lg._id,
                   lg._filename.c_str(),
                   lg._filenumber,
                   lg._curr_time.c_str(),
                   lg._message_info.c_str());
        }
        void FlushLogToFile(const logmessage &lg)
        {
            std::ofstream out(_logfile, std::ios::app);
            if (!out.is_open())
                return;
            char logtxt[2048];
            snprintf(logtxt, sizeof logtxt, "[%s][%d][%s][%d][%s] %s",
                     lg._level.c_str(),
                     lg._id,
                     lg._filename.c_str(),
                     lg._filenumber,
                     lg._curr_time.c_str(),
                     lg._message_info.c_str());
            out.write(logtxt, strlen(logtxt));
            out.close();
        }
        void FlushLog(const logmessage &lg)
        {
            LockGuard lockguard(&glock);
            switch (_type)
            {
            case SCREEN_TYPE:
                FlushLogToScreen(lg);
                break;
            case FILE_TYPE:
                FlushLogToFile(lg);
                break;
            }
        }
        void logMessage(std::string filename, int filenumber, int level, const char *format, ...)
        {
            logmessage lg;

            lg._level = LevelToString(level);
            lg._id = getpid();
            lg._filename = filename;
            lg._filenumber = filenumber;
            lg._curr_time = GetCurrTime();

            va_list ap;
            va_start(ap, format);
            char log_info[1024];
            vsnprintf(log_info, sizeof(log_info), format, ap);
            va_end(ap);
            lg._message_info = log_info;

            // 打印出来日志
            FlushLog(lg);
        }
        ~log()
        {
        }

    private:
        int _type;
        std::string _logfile;
    };

    log lg;

#define LOG(level, Format, ...)                                        \
    do                                                                 \
    {                                                                  \
        lg.logMessage(__FILE__, __LINE__, level, Format, ##__VA_ARGS__); \
    } while (0)
#define EnableScreen()          \
    do                          \
    {                           \
        lg.Enable(SCREEN_TYPE); \
    } while (0)
#define EnableFile()          \
    do                        \
    {                         \
        lg.Enable(FILE_TYPE); \
    } while (0)
};