Linux|多线程（三）

线程池

线程池是一种多线程处理形式，处理过程中它将被提交的任务分配给预先创建好的多个线程中的一个去执行。

线程池的实现

#pragma once
#include <pthread.h>
#include <vector>
#include <string>
#include <unistd.h>
#include <pthread.h>
#include <queue>
#include <functional>
#include "task.hpp"
#include "thread.hpp"
using namespace ThreadModel;
// void test(const std::string &name)
// {

//     while (1)
//     {
//         std::cout << name << "is running.." << std::endl;
//         sleep(1);
//     }
// }
const int Default_nums = 3;

template <class T>
class ThreadPool
{
    using func_t = std::function<void(const std::string name)>;

private:
    bool isEmpty()
    {
        return _task_queue.size() == 0;
    }
    void Sleep()
    {
        pthread_cond_wait(&_cond, &_mutex);
    }
    void lockQueue()
    {
        pthread_mutex_lock(&_mutex);
    }
    void wakeUp()
    {
        pthread_cond_signal(&_cond);
    }
    void wakeUpAll()
    {
        pthread_cond_broadcast(&_cond);
    }

    void unlockQueue()
    {
        pthread_mutex_unlock(&_mutex);
    }
    void hadertask(const std::string name)
    {
        lockQueue();
        while (true)
        {
            while (isEmpty() && _isrunning)
            {
                _sleepnums++;
                std::cout << name << " star sleep" << std::endl;
                Sleep();
                std::cout << name << " awaken" << std::endl;
                _sleepnums--;
            }
            if (isEmpty() && !_isrunning)
            {
                unlockQueue();
                std::cout << name << " quit" << std::endl;
                break;
            }
            T t = _task_queue.front();
            _task_queue.pop();
            unlockQueue();
            t();
            std::cout << t.solve() << std::endl;
        }
    }
    void init()
    {
        // 并将调用func时的第一个参数传递给hander函数。
        func_t func = std::bind(&ThreadPool::hadertask, this, std::placeholders::_1);
        for (int i = 0; i < _thread_nums; i++)
        {

            std::string name = "thread-" + std::to_string(i + 1);

            _threads.emplace_back(name, func);
            std::cout << name << " init sucess" << std::endl;
        }
    }
    ThreadPool(int thread_nums = Default_nums) : _thread_nums(thread_nums), _sleepnums(0)
    {
        init();
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
    }

public:
    ~ThreadPool()
    {
        pthread_cond_destroy(&_cond);
        pthread_mutex_destroy(&_mutex);
    }
    static ThreadPool<T> * GetInstance(int thread_nums = Default_nums)
    {
        if (_tp == nullptr)
        {
            pthread_mutex_lock(&_sig_mutx);
            if (_tp == nullptr)
            {
                _tp = new ThreadPool(thread_nums);
                std::cout<<"create instance"<<std::endl;
            }
            pthread_mutex_unlock(&_sig_mutx);
        }
        std::cout<<"get instance"<<std::endl;
        return _tp;
    }

    void start()
    {
        _isrunning = true;
        std::cout << "threadpool is running:" << std::endl;

        for (auto &thread : _threads) // 这个引用特别重要
        // 如果不传引用 thread 出了函数会被销毁 访问name 会出问题
        {

            thread.start();
        }
    }

    void equeue(const T &in)
    {
        // 拿数据和放数据要加锁 因为可能脏读
        lockQueue();
        if (_isrunning)
        {
            _task_queue.push(in);
            if (_sleepnums > 0)
            {
                wakeUp();
            }
        }

        std::cout << in.inquiry() << std::endl;

        unlockQueue();
    }
    void stop()
    {
        wakeUpAll();
        std::cout << "threadpool stop" << std::endl;
        _isrunning = false;
    }

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

    pthread_mutex_t _mutex;
    pthread_cond_t _cond;

    static ThreadPool *_tp;
    static pthread_mutex_t _sig_mutx;
};
template <class T>
ThreadPool<T> *ThreadPool<T>::_tp = nullptr;

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

日志类

#pragma once
#include <iostream>
#include <string>
#include <unistd.h>
#include <sys/types.h>
#include <ctime>
#include <stdarg.h>
#include <fstream>
#include <cstring>
#include <pthread.h>
#include "lockerguard.hpp"
// namespace{}
namespace log_ns
{
#define SCREEN_TYPE 1
#define FILE_TYPE 2
    pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;
    enum level
    {
        DEBUG = 1,
        INFO,
        WARNNING,
        ERROR,
        FATAL
    };
    std::string levelToString(int level)
    {
        switch (level)
        {
        case DEBUG:
            return "DEBUG";
        case INFO:
            return "INFO";
        case WARNNING:
            return "WARNNING";
        case ERROR:
            return "ERROR";
        case FATAL:
            return "FATAL";
        default:
            return "UNKNOW";
        }
    }
    std::string GetCurTime()
    {
        time_t now = time(nullptr);
        struct tm *cur_time = localtime(&now);
        char buff[128];
        snprintf(buff, 128, "%d-%02d-%02d %02d:%02d:%02d",
                 cur_time->tm_year + 1900,
                 cur_time->tm_mon + 1,
                 cur_time->tm_mday,
                 cur_time->tm_hour,
                 cur_time->tm_min,
                 cur_time->tm_sec);
        return buff;
    }
    class LogMessage
    {
    public:
        std::string _level;        // 日志等级
        pid_t _id;                 // 进程号
        std::string _filename;     // 文件名
        int _filenumber;           // 行号
        std::string _cur_time;     // 日志所写时间
        std::string _message_info; // 日志信息
    };
    const char *gfile = "./log.txt";
    class Log
    {
    public:
        Log(const std::string &logfile = gfile) : _logFile(logfile), _type(SCREEN_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._cur_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 buff[1024];
            snprintf(buff, sizeof(buff), "[%s][%d][%s][%d][%s] %s",
                     lg._level.c_str(),
                     lg._id,
                     lg._filename.c_str(),
                     lg._filenumber,
                     lg._cur_time.c_str(),
                     lg._message_info.c_str());
            out.write(buff, strlen(buff));
            out.close();
        }
        void FlushLog(const LogMessage &lg)
        {
            LockGuard lockguard(&gmutex);
            switch (_type)
            {
            case SCREEN_TYPE:
                FlushLogToScreen(lg);
                break;
            case FILE_TYPE:
                FlushLogToFile(lg);
                break;
            default:
                break;
            }
        }
        void logMessage(int level, std::string filename, int filenumber, const char *format...)
        {
            LogMessage lg;
            lg._level = levelToString(level);
            lg._filename = filename;
            lg._filenumber = filenumber;
            lg._cur_time = GetCurTime();
            lg._id = getpid();
            va_list ap;
            va_start(ap, format);
            char buff[128];
            vsnprintf(buff, sizeof(buff), format, ap);
            va_end(ap);
            FlushLog(lg);
        }

    private:
        int _type;
        std::string _logFile;
    };
    Log lg;
#define LOG(Level, Format, ...)                                          \
    do                                                                   \
    {                                                                    \
        lg.logMessage(Level, __FILE__, __LINE__, Format, ##__VA_ARGS__); \
    } while (0)
#define EnableScreen()          \
    do                          \
    {                           \
        lg.Enable(SCREEN_TYPE); \
    } while (0)
#define EnableFile()          \
    do                        \
    {                         \
        lg.Enable(FILE_TYPE); \
    } while (0)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

单例模式

 static ThreadPool<T> * GetInstance(int thread_nums = Default_nums)
    {
        if (_tp == nullptr)
        {
            pthread_mutex_lock(&_sig_mutx);
            if (_tp == nullptr)
            {
                _tp = new ThreadPool(thread_nums);
                std::cout<<"create instance"<<std::endl;
            }
            pthread_mutex_unlock(&_sig_mutx);
        }
        std::cout<<"get instance"<<std::endl;
        return _tp;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15