胧の树洞

Mutex.hpp

#pragma once

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


class Mutex
{
public:
    Mutex()
    {
        pthread_mutex_init(&_lock , nullptr);
    }

    void Lock()
    {
        pthread_mutex_lock(&_lock);
    }

    pthread_mutex_t *Ptr()
    {
        return &_lock;
    }

    void Unlock()
    {
        pthread_mutex_unlock(&_lock);
    }

    ~Mutex()
    {
        pthread_mutex_destroy(&_lock);
    }
private:
    pthread_mutex_t _lock;
};

class LockGuard
{
public:
    LockGuard(Mutex &lock):_lockref(lock)
    {
        _lockref.Lock();
    }

    ~LockGuard()
    {
        _lockref.Unlock();
    }
private:
    Mutex &_lockref;
};

‍

Cond.hpp

#ifndef __COND_HPP
#define __COND_HPP

#include <pthread.h>
#include "Mutex.hpp"

class Cond
{
public:
    Cond()
    {
        pthread_cond_init(&_cond, nullptr);
    }
    void Wait(Mutex &mutex)
    {
        int n = pthread_cond_wait(&_cond, mutex.Ptr());
        (void)n;
    }
    void Signal()
    {
        int n = pthread_cond_signal(&_cond);
        (void)n;
    }
    void Broadcast()
    {
        int n = pthread_cond_broadcast(&_cond);
        (void)n;
    }
    ~Cond()
    {
        pthread_cond_destroy(&_cond);
    }
private:
    pthread_cond_t _cond;
};

#endif

Thread.hpp

#pragma once

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

namespace NS_THREAD_MODULE
{
    static int gnumber = 1;
    using callback_t = std::function<void()>;

    enum class TSTATUS
    {
        THREAD_NEW,
        THREAD_RUNNING,
        THREAD_STOP
    };

    std::string Status2String(TSTATUS s)
    {
        switch (s)
        {
        case TSTATUS::THREAD_NEW:
            return "THREAD_NEW";
        case TSTATUS::THREAD_RUNNING:
            return "THREAD_RUNNING";
        case TSTATUS::THREAD_STOP:
            return "THREAD_STOP";
        default:
            return "UNKNOWN";
        }
    }

    std::string IsJoined(bool joinable)
    {
        return joinable ? "true" : "false";
    }

    class Thread
    {
    private:
        void ToRunning()
        {
            _status = TSTATUS::THREAD_RUNNING;
        }
        void ToStop()
        {
            _status = TSTATUS::THREAD_STOP;
        }
        static void *ThreadRoutine(void *args)
        {
            Thread *self = static_cast<Thread *>(args);
            pthread_setname_np(self->_tid, self->_name.c_str());
            self->_cb();
            self->ToStop();
            return nullptr;
        }

    public:
        Thread(callback_t cb)
            : _tid(-1), _status(TSTATUS::THREAD_NEW), _joinable(true), _cb(cb), _result(nullptr)
        {
            _name = "New-Thread-" + std::to_string(gnumber++);
        }
        bool Start()
        {
            int n = pthread_create(&_tid, nullptr, ThreadRoutine, this);
            if (n != 0)
                return false;

            ToRunning();
            return true;
        }
        void Join()
        {
            if (_joinable)
            {
                int n = pthread_join(_tid, &_result);
                if (n != 0)
                {
                    std::cerr << "join error: " << n << std::endl;
                    return;
                }
                (void)_result;
                _status = TSTATUS::THREAD_STOP;
            }
            else
            {
                std::cerr << "error, thread join status: " << IsJoined(_joinable) << std::endl;
            }
        }
        // 暂停
        // void Stop() // restart()
        // {
        //     // 让线程暂停
        // }
        void Die()
        {
            if (_status == TSTATUS::THREAD_RUNNING)
            {
                pthread_cancel(_tid);
                _status = TSTATUS::THREAD_STOP;
            }
        }
        void Detach()
        {
            if (_status == TSTATUS::THREAD_RUNNING && _joinable)
            {
                pthread_detach(_tid);
                _joinable = false;
            }
            else
            {
                std::cerr << "detach " << _name << " failed" << std::endl;
            }
        }
        void PrintInfo()
        {
            std::cout << "thread name : " << _name << std::endl;
            std::cout << "thread _tid : " << _tid << std::endl;
            std::cout << "thread _status : " << Status2String(_status) << std::endl;
            std::cout << "thread _joinable : " << IsJoined(_joinable) << std::endl;
        }

        ~Thread()
        {
        }

    private:
        std::string _name;
        pthread_t _tid;
        TSTATUS _status;
        bool _joinable;
        // 线程要有自己的任务处理，即回调函数
        callback_t _cb;

        // 线程退出信息
        void *_result;
    };
}

Logger.hpp

#ifndef __LOGGER_HPP
#define __LOGGER_HPP

#include<iostream>
#include<cstdio>
#include<string>
#include<memory> 
#include<sstream>
#include<ctime>
#include<sys/time.h>
#include<unistd.h>
#include<filesystem>
#include<sys/stat.h>
#include<fstream>
#include"Mutex.hpp"

namespace NS_LOG_MODULE
{
    enum class LogLevel
    {
        INFO,
        WARNING,
        ERROR,
        FATAL,
        DEBUG
    };

    std::string LogLevel2Message(LogLevel level)
    {
        switch(level)
        {
            case LogLevel::INFO:
                    return "INFO";
            case LogLevel::WARNING:
                    return "WARNING";
            case LogLevel::ERROR:
                    return "ERROR";
            case LogLevel::FATAL:
                    return "FATAL";
            case LogLevel::DEBUG:
                    return "DEBUG";
            default:
                    return "UNKNOWN";
        }
    }

    std::string GetCurrentTime()
    {
        struct timeval current_time;
        int n = gettimeofday(&current_time , nullptr);
        (void)n;

        struct tm struct_time;
        localtime_r(&(current_time.tv_sec) , &struct_time);

        char timestr[128];
        snprintf(timestr , sizeof(timestr) , "%04d-%02d-%02d %02d:%02d:%02d.%ld" , 
                struct_time.tm_year + 1900,
                struct_time.tm_mon + 1,
                struct_time.tm_mday,
                struct_time.tm_hour,
                struct_time.tm_min,
                struct_time.tm_sec,
                current_time.tv_usec);
        return timestr;
    }

    // 策略模式，策略接口
    // 1. 显示器打印
    // 2. 文件写入
    class LogStrategy
    {
    public:
        virtual ~LogStrategy() = default;
        virtual void SyncLog(const std::string &message) = 0;

    };

    // 控制台日志刷新策略, 日志将来要向显示器打印
    class ConsoleStartegy : public LogStrategy 
    {
    public:
        void SyncLog(const std::string &message) override
        {
            LockGuard lockguard(_mutex);
            std::cerr <<message <<"\n";
        } 

        ~ConsoleStartegy()
        {}

    private:
        Mutex _mutex;
    };


    const std::string defaultpath = "./log";
    const std::string defaultname = "log.txt";

    // 文件策略
    class FileLogStrategy :public LogStrategy
    {
    public:
        FileLogStrategy(const std::string &path = defaultpath , const std::string &name = defaultname):
                        _logpath(path),
                        _logfilename(name)
        {
            if(std::filesystem::exists(_logpath))
                return;
            try
            {
                std::filesystem::create_directories(_logpath);
            }
            catch(const std::filesystem::filesystem_error &e)
            {
                std::cerr<<e.what() <<"\n";
            }

        }


        void SyncLog(const std::string &message) override
        {
            LockGuard lockguard(_mutex);
            if(!_logpath.empty() && _logpath.back() !='/')
            {
                _logpath += "/";
            }
            std::string targetlog = _logpath + _logfilename;
            std::ofstream out(targetlog , std::ios::app); // 追加写入
            if(!out.is_open())
            {
                std::cerr<<"open" <<targetlog <<"failen" <<std::endl;
                return;
            }
            out << message << "\n";
            out.close();
        }

        ~FileLogStrategy()
        {}

    private:
        std::string _logpath;
        std::string _logfilename;
        Mutex _mutex;
    };

    class Logger
    {
        // 日志生成



    public:
        Logger()
        {
            UseConsoleStrategy();
        }

        void UseConsoleStrategy()
        {
            _strategy = std::make_unique<ConsoleStartegy>();
        }

        void UseFileStrategy()
        {
            _strategy = std::make_unique<FileLogStrategy>();
        }

        // void Debug(const std::string &message)
        // {
        //     if(_strategy != nullptr)
        //     {
        //         _strategy->SyncLog(message);
        //     }
        // }

        class LogMessage
        {
        public:
            LogMessage(LogLevel level , std::string &filename , int line , Logger &logger):
            _level(level),
            _curr_time(GetCurrentTime()),
            _pid(getpid()),
            _filename(filename),
            _line(line),
            _logger(logger)
            {
                std::stringstream ss;
                ss  <<"[" <<_curr_time <<"] " 
                    <<"[" <<LogLevel2Message(_level) <<"] "
                    <<"[" <<_pid <<"] " 
                    <<"[" <<_filename <<"] "
                    <<"[" <<_line <<"] "
                    <<" : ";    

                    _loginfo = ss.str();
            }

            template<typename T>
            LogMessage &operator << (const T &info)
            {
                std::stringstream ss;
                ss <<info;
                _loginfo += ss.str();
                return *this; // 返回当前LogMessage对象
            }

            ~LogMessage()
            {
                if(_logger._strategy)
                {
                    _logger._strategy->SyncLog(_loginfo);
                }
            }

        private:
            LogLevel _level;
            std::string _curr_time;
            pid_t _pid;
            std::string _filename;
            int _line;
            std::string _loginfo;
            Logger &_logger;
        };

        LogMessage operator()(LogLevel level , std::string filename , int line)
        {
            return LogMessage(level , filename , line , *this);
        }

        ~Logger()
        {}

    private:
        std::unique_ptr<LogStrategy> _strategy;  // 刷新策略
    };

Logger logger;

#define ENABLE_CONSOLE_LOG_STRATEGY() logger.UseConsoleStrategy();
#define ENABLE_FILE_LOG_STRATEGY() logger.UseFileStrategy();


#define LOG(level) logger(level , __FILE__ , __LINE__)


}
#endif

ThreadPool.hpp

#pragma once

#include <iostream>
#include <vector>
#include <queue>
#include "Logger.hpp"
#include "Thread.hpp"
#include "Cond.hpp"

namespace NS_THREAD_POOL
{
    using namespace NS_LOG_MODULE;
    using namespace NS_THREAD_MODULE;

    const int defaultnum = 5;

    // void Test()
    // {
    //     char name[128];
    //     pthread_getname_np(pthread_self(), name, sizeof(name));
    //     while ((true))
    //     {
    //         LOG(LogLevel::DEBUG) << "运行线程" << name;
    //         sleep(1);
    //     }
    // }

    template <typename T>
    class ThreadPool
    {
    private:
        void HendlerTask()
        {
            char name[128];
            pthread_getname_np(pthread_self(), name, sizeof(name));
            while ((true))
            {
                T task;
                {
                    LockGuard lockguard(_mutex);

                    while (_tasks.empty() && _isrunning)
                    {
                        _slaver_sleeper_count++;
                        _cond.Wait(_mutex);
                        _slaver_sleeper_count--;
                    }

                    if (!_isrunning && _tasks.empty())
                    {
                        _mutex.Unlock();
                        break;
                    }

                    task = _tasks.front();
                    _tasks.pop();
                }
                LOG(LogLevel::INFO) << name << "处理任务：";
                task();
                LOG(LogLevel::DEBUG) << task.Result();
            }
            LOG(LogLevel::INFO) << name << "quit...";
        }

        ThreadPool(int slaver_num = defaultnum) : _isrunning(false), _slaver_sleeper_count(0), _slaver_num(slaver_num)
        {
            for (int idx = 0; idx < _slaver_num; idx++)
            {
                // auto f = std::bind(&ThreadPool::HendlerTask , this);
                // auto f = [this](){
                //     this->HendlerTask();
                // };
                // _slavers.emplace_back(f);

                _slavers.emplace_back([this]()
                                      { this->HendlerTask(); });
            }
        }

        // 赋值 拷贝构造禁止
        ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
        ThreadPool(const ThreadPool<T> &) = delete;

    public:
        static ThreadPool<T> *Instance()
        {
            if(nullptr == _instance)
            {
                LockGuard lockguard(_lock);
                if (nullptr == _instance)
                {
                    _instance = new ThreadPool<T>();
                    _instance->Start();
                    LOG(LogLevel::INFO) << "创建线程池对象";
                }
            }
            return _instance;
        }
        void Start()
        {
            if (_isrunning)
            {
                LOG(LogLevel::WARNING) << "Thread Pool Is Already Running";
                return;
            }
            _isrunning = true;
            for (auto &slave : _slavers)
            {
                slave.Start();
            }
        }

        void Stop()
        {
            _mutex.Lock();
            _isrunning = false;
            if (_slaver_sleeper_count > 0)
                _cond.Broadcast();
            _mutex.Unlock();
        }

        void Wait()
        {
            for (auto &slave : _slavers)
            {
                slave.Join();
            }
        }

        void EnQueue(T in)
        {
            _mutex.Lock();
            _tasks.push(in);
            if (_slaver_sleeper_count > 0)
                _cond.Signal();
            _mutex.Unlock();
        }

        ~ThreadPool()
        {
        }

    private:
        bool _isrunning;
        int _slaver_num;
        std::vector<Thread> _slavers;
        std::queue<T> _tasks;
        Mutex _mutex;
        Cond _cond;
        int _slaver_sleeper_count;

        static ThreadPool<T> *_instance;
        static Mutex _lock;
    };

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

    template <typename T>
    Mutex ThreadPool<T>::_lock;
}

Main.cc

#include"Logger.hpp"
#include"ThreadPool.hpp"

#include<iostream>
#include<memory>
#include<functional>
#include<ctime>
#include<cstdlib>

using namespace NS_LOG_MODULE;
using namespace NS_THREAD_POOL;

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

class Task
{
public:
    Task(){}
    Task(int x , int y): _x(x) , _y(y)
    {}

    void operator()()
    {
        _result = _x + _y;
    }

    std::string Result()
    {
        return std::to_string(_x) + " + " + std::to_string(_y) +" = " + std::to_string(_result);
    }

    ~Task(){}

private:
    int _x;
    int _y;
    int _result;
};

int main()
{

    ENABLE_CONSOLE_LOG_STRATEGY();
    srand((long)time(nullptr) ^ getpid());
    int cnt = 10;

    sleep(5);

    while(cnt--)
    {
        int x = rand() % 11 + 1;
        usleep(337);
        int y = rand() % 13 + 1;
        Task t(x , y);
        ThreadPool<Task>::Instance()->EnQueue(t);
        sleep(1);
    }

    ThreadPool<Task>::Instance()->Stop();
    ThreadPool<Task>::Instance()->Wait();

    return 0;
}

Makefile

threadpool:Main.cc
	g++ -o $@ $^ -std=c++17 -g
.PHONY:clean
clean:
	rm -f threadpool

‍