抱歉，您的浏览器无法访问本站

本页面需要浏览器支持（启用）JavaScript


了解详情 >

线程池

秃头王
OOP

发布于：2022年7月26日
次浏览

多线程程序设计

功能函数和线程入口函数要分开就会满足高内聚低耦合的思想

临界资源：大家都能访问到的资源

1
2
3
4
5
6
7
8
// 全局变量 int ans = 0; 、std::mutex m_mutex;
// 加锁的地方就是临界区
// 抢占胡互斥锁, 如果抢占上了的话才对ans + 1, 出了作用域就会呗释放
unique_lock<mutex> lock(m_mutex);
ans += is_prime(i);
lock.unlock();
// 对这种加操作的时候,有可能会发生多线程抢占资源的时候的时候导致数据不正确
// 临界资源
1
g++ **.cpp -pthread

500W个数查看素数个数

  • prime_count_test_15s::main();
  • prime_count_test1_2m10s::main();
  • prime_count_test2_14s::main();
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
/*************************************************************************
        > File Name: 1.thredd_pool.cpp
        > Author: 秃头王
        > Mail: 1658339000@qq.com
        > Created Time: 2022年07月25日 星期一 18时08分16秒
 ************************************************************************/

#include <iostream>
#include <thread>
using namespace std;

#define BEGINS(x) namespace x {
#define ENDS(x) }

BEGINS(thread_usage)

void func() {
    cout << "hello world" << endl;
}

void print(int a, int b) {
    cout << a << " " << b << endl;
    return ;
}

int main() {
    // 线程
    thread t1(func);
    t1.join();
    thread t2(print, 1, 2);
    t2.join();
    return 0;
}

ENDS(thread_usage)

BEGINS(prime_count_test_15s)

bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

int prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}

// 给多线程执行的入口函数
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin" << endl;
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "done" << endl;

}

int main() {
    #define batch 5000000
    thread *t[10];
    int ret[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(worker, j, j + batch - 1, ref(ret[i]));
    }

    for (auto x : t) x->join();
    int ans = 0;
    for(auto x : ret) ans += x;
    cout << ans << endl;
    for(auto x : t) delete x;
    #undef batch
    return 0;
}

ENDS(prime_count_test)

#include <mutex>

BEGINS(prime_count_test1_2m10s)

int ans = 0;
std::mutex m_mutex;


bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

void prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        // 加锁的地方就是临界区
        // 抢占胡互斥锁, 如果抢占上了的话才对ans + 1, 出了作用域就会呗释放
        unique_lock<mutex> lock(m_mutex);
        ans += is_prime(i);
        lock.unlock();
        // 对这种加操作的时候,有可能会发生多线程抢占资源的时候的时候导致数据不正确
        // 临界资源
    }
    return ;
}


int main() {
    #define batch 5000000
    thread *t[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(prime_count, j, j + batch - 1);
    }

    for (auto x : t) x->join();
    for(auto x : t) delete x;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(prime_count_test1)

// 第二种加锁
BEGINS(prime_count_test2_14s)

int ans = 0;
std::mutex m_mutex;


bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

void prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        int ret = is_prime(i);
        __sync_fetch_and_add(&ans, ret);
    }
    return ;
}


int main() {
    #define batch 5000000
    thread *t[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(prime_count, j, j + batch - 1);
    }

    for (auto x : t) x->join();
    for(auto x : t) delete x;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(prime_count_test1)

int main() {
    // thread_usage::main();
    // prime_count_test_15s::main();
    // prime_count_test1_2m10s::main();
    prime_count_test2_14s::main();
    return 0;
}

线程池(简版)-代码实现

进程：进程是资源分配的最基本单位

进程申请一份线程就占8m的空间，进程又申请一份线程又占8m的空间

image-20220726160441707

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
/*************************************************************************
        > File Name: 1.thredd_pool.cpp
        > Author: 秃头王
        > Mail: 1658339000@qq.com
        > Created Time: 2022年07月25日 星期一 18时08分16秒
 ************************************************************************/

#include <iostream>
#include <thread>
using namespace std;
#include <vector>
#include <unordered_map>
#include <queue>
#include <condition_variable>

BEGINS(thread_pool_test)

// 函数方法绑定到一起的
class Task {
public :

    template<typename FUN_T, typename ...ARGS>
    Task(FUN_T func, ARGS... args) {
        // 用forward向下传时候可以保证准确的值传递
        this->func = bind(func, forward<ARGS>(args)...);
        // 右值传递时候可能就会出错
        // this->func = bind(func, fargs...);
    }

    void run() {
        func();
        return ;
    }

private :
    function<void()> func;
};


// 线程池类
class ThreadPool {
public :
    ThreadPool(int n = 1) : thread_size(n), threads(n), starting(false) {
        this->start();
    }

    // 取任务、执行任务
    void worker() {
        auto id = this_thread::get_id();
        running[id] = true;
        while(running[id]) {
            Task *t = get_task();
            t->run();
            delete t;
        }
    }

    void start() {
        if(starting == true) return ;
        for(int i = 0; i < thread_size; i++) {
            threads[i] = new thread(&ThreadPool::worker, this);
        }
        starting = true;
        return ;
    }

    void stop() {
        if(starting == false) return ;
        // 添加毒药任务
        for(int i = 0; i < threads.size(); i++) {
            this->add_task(&ThreadPool::stop_running, this);
        }
        for(int i = 0; i < threads.size(); i++) {
            threads[i]->join();
        }
        for(int i = 0; i < threads.size(); i++) {
            delete threads[i];
            threads[i] = nullptr;
        }
        starting = false;
        return ;
    }

    template<typename FUNC_T, typename ...ARGS>
    void add_task(FUNC_T func, ARGS... args) {
        unique_lock<mutex> lock(m_mutex);
        tasks.push(new Task(func, forward<ARGS>(args)...));

        // 通知所有等待着任务的线程
        m_cond.notify_one();
        return ;
    }

    ~ThreadPool() {
        this->stop();
        while(!tasks.empty()) {
            delete tasks.front();
            tasks.pop();
        }
        return ;
    }

private :
    Task *get_task() {
       unique_lock<mutex> lock(m_mutex);
        // 防止虚假唤醒 wait 会自动释放锁
        while (tasks.empty()) m_cond.wait(lock);
        Task *t = tasks.front();
        tasks.pop();
        return t;
    }

    // 毒药任务
    void stop_running() {
        auto id = this_thread::get_id();
        running[id] = false;
        return ;
    }

    bool starting;
    int thread_size;
    vector<thread *> threads;
    unordered_map<decltype(this_thread::get_id()), bool> running;
    queue<Task *> tasks;

    std::condition_variable m_cond;
    std::mutex m_mutex;

};

bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

int prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}

// 给多线程执行的入口函数
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin" << endl;
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "done" << endl;

}

int main() {
    ThreadPool tp(5);
     #define batch 5000000
    thread *t[10];
    int ret[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        tp.add_task(worker, j, j + batch - 1, ref(ret[i]));
    }
    tp.stop();
    int ans = 0;
    for(auto x : ret) ans +=x ;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(thread_pool_test)

int main() {
    thread_pool_test::main();
    return 0;
}

线程池(扩充)-代码实现

增加:

  • 增加: 线程日志模块
    • threadpoolPlus.h

threadpoolPlus.h

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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
/*************************************************************************
	> File Name: threadpoolPlus.h
	> Author: 秃头王
	> Mail: 1658339000@qq.com
	> Created Time: 2022年07月27日 星期三 15时37分13秒
 ************************************************************************/

#ifndef _THREADPOOLPLUS_H
#define _THREADPOOLPLUS_H

#include <iostream>
#include <thread>
#include <mutex>
#include <queue>
#include <vector>
#include <unordered_map>
#include <condition_variable>
#include <functional>
using namespace std;

#define BEGINS(x) namespace x {
#define ENDS(x) }

BEGINS(ttw)

// 函数方法绑定到一起的
class Task {
public :

    template<typename FUN_T, typename ...ARGS>
    Task(FUN_T func, ARGS... args);
    void run();

private :
    function<void()> func;
};

template<typename FUN_T, typename ...ARGS>
Task::Task(FUN_T func, ARGS... args) {
    // 用forward向下传时候可以保证准确的值传递
    this->func = bind(func, forward<ARGS>(args)...);
    // 右值传递时候可能就会出错
    // this->func = bind(func, fargs...);
}

void Task::run() {
    Task::func();
    return ;
}

// END Task

// 线程池类
class ThreadPool {
public :
    ThreadPool(int n = 1);

    // 取任务、执行任务
    void worker();
   
    void start();

    void stop();
    
    template<typename FUNC_T, typename ...ARGS>
    void add_task(FUNC_T, ARGS...);
    
    ~ThreadPool();

private :
    // 获取任务
    Task *get_task();
    // 毒药任务
    void stop_running();

    bool starting;
    int thread_size;
    vector<thread *> threads;
    unordered_map<decltype(this_thread::get_id()), bool> running;
    queue<Task *> tasks;

    std::condition_variable m_cond;
    std::mutex m_mutex;
    
};

ThreadPool::ThreadPool(int n) : thread_size(n), threads(n), starting(false) {
    this->start();
}

// 取任务、执行任务
void ThreadPool::worker() {
    auto id = this_thread::get_id();
    running[id] = true;
    while(running[id]) {
        Task *t = get_task();
        t->run();
        delete t;
    }
    return ;
}

void ThreadPool::start() {
    if(starting == true) return ;
    for(int i = 0; i < thread_size; i++) {
        threads[i] = new thread(&ThreadPool::worker, this);
    }
    starting = true;
    return ;
}

void ThreadPool::stop() {
    if(starting == false) return ;
    for(int i = 0; i < threads.size(); i++) {
        this->add_task(&ThreadPool::stop_running, this);
    }
    for(int i = 0; i < threads.size(); i++) {
        threads[i]->join();
    }
    for(int i = 0; i < threads.size(); i++) {
        delete threads[i];
        threads[i] = nullptr;
    }
    starting = false;
    return ; 
}

template<typename FUNC_T, typename ...ARGS>
void ThreadPool::add_task(FUNC_T func, ARGS... args) {
    unique_lock<mutex> lock(m_mutex);
    tasks.push(new Task(func, forward<ARGS>(args)...));

    // 通知所有等待着任务的线程
    m_cond.notify_one();
    return ;
}

ThreadPool::~ThreadPool() {
    this->stop();
    while(!tasks.empty()) {
        delete tasks.front();
        tasks.pop();
    }
    return ;
}

// ThreadPool private begin
Task *ThreadPool::get_task() {
    unique_lock<mutex> lock(m_mutex);
    // 防止虚假唤醒 wait 会自动释放锁
    while (tasks.empty()) m_cond.wait(lock);
    Task *t = tasks.front();
    tasks.pop();
    return t;
}

void ThreadPool::stop_running() {
    auto id = this_thread::get_id();
    running[id] = false;
    return ;
}
// ThreadPool private end


// END ThreadPool

bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false; 
    }
    return true;
}

int prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}

// 给多线程执行的入口函数
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin" << endl;
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "done" << endl;

}

int main() {
    ThreadPool tp(5);
     #define batch 5000000
    thread *t[10];
    int ret[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        tp.add_task(worker, j, j + batch - 1, ref(ret[i]));
    }
    tp.stop();
    int ans = 0;
    for(auto x : ret) ans +=x ;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(ttw)

#endif

logger.cpp

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
/*************************************************************************
	> File Name: logger.cpp
	> Author: 秃头王
	> Mail: 1658339000@qq.com
	> Created Time: 2022年07月27日 星期三 15时38分31秒
 ************************************************************************/

/*
 *  cout << "hellow world" << 123 << endl;
 *  当多个线程都在操作时候有的在打印hello world有的在打印123
 *  如果把这这个语句变成原子操作
 *  原子操作：不可分割的一个单独的操作
 *  原子操作必加锁、原子操作你不加锁在作死
 *  
 *  cout << "hellow world" << 123 << endl; -> 对应了一个单独的缓冲区
 *  cout << "hellow world" << 123 << endl; -> 对应了一个单独的缓冲区
 *  cout << "hellow world" << 123 << endl; -> 对应了一个单独的缓冲区 -> 对象
 *  在现在这个场景种 cout 对应的都是一个所以不安全
 *
 *  用到临时对象
 */

/*
    std::ostringstream s;
    s << "hellow world" << " " << 123 << std::endl;
    std::cout << s.str();
*/

#include <iostream>
#include <sstream>
#include <string>
#include <map>
#include <mutex>
#include "threadpoolPlus.h"

/*
#define LEVEL(str) ttw::LogLevel::##str
#define LEVEL_INFO    LEVEL(INFO)
#define LEVEL_WARNING LEVEL(WARNING)
#define LEVEL_DEBUG   LEVEL(DEBUG)
#define LEVEL_ERROR   LEVEL(ERROR)
#define LEVEL_FATAL   LEVEL(FATAL)
*/

// __FILE__(文件名) __LINE__(行哈) level(日志等级)
#define LOG(level) ttw::Logger::LoggerStream(level, __FILE__, __LINE__, ttw::ttw_log)
#define LOG_INFO LOG(ttw::LogLevel::INFO)
#define LOG_WARNING LOG(ttw::LogLevel::WARNING)
#define LOG_DEBUG LOG(ttw::LogLevel::DEBUG)
#define LOG_ERROE LOG(ttw::LogLevel::ERROR)
#define LOG_FATAL LOG(ttw::LogLevel::FATAL)
#define SET_LEVEL(level) ttw::ttw_log.set_level(level)

namespace ttw {
    // 枚举类
    class LogLevel {
    public :
        static const int INFO ;
        static const int WARNING;
        static const int DEBUG;
        static const int ERROR;
        static const int FATAL;
    };
    const int LogLevel::INFO = 1;
    const int LogLevel::WARNING = 2;
    const int LogLevel::DEBUG = 3;
    const int LogLevel::ERROR = 4;
    const int LogLevel::FATAL = 5;
    std::map<int, std::string> LevelString = {
        {LogLevel::INFO, "INFO"},
        {LogLevel::WARNING, "WARNING"},
        {LogLevel::DEBUG, "DEBUG"},
        {LogLevel::ERROR, "ERROR"},
        {LogLevel::FATAL, "FATAL"}
    };

    class Logger {
    public :
        Logger() : LOG_LEVEL(LogLevel::INFO) {}
        // 临时对象
        class LoggerStream : public std::ostringstream {
        public :
            LoggerStream(int level, const char *file_name, int line_no, Logger &raw_log) : line_no(line_no),level(level), raw_log(raw_log) {
                std::ostringstream &now = *this;
                now << "[" << file_name << " : " << LevelString[level] << "]"; 
            }
            ~LoggerStream() {
                if(level < raw_log.LOG_LEVEL) return ;
                // 互斥锁
                std::unique_lock<std::mutex> lock(raw_log.m_mutex);
                std::cout << this->str() << "(" << line_no << ")" << std::endl;
            }
        private :
            // 行号、日志等级
            int line_no, level;
            Logger &raw_log;
        };
        void set_level(int level) { this->LOG_LEVEL = level; }
        int LOG_LEVEL;
        std::mutex m_mutex;
    };
    // 输出流对象
    Logger ttw_log;

} // end of ttw


void func(int a, int b, int c) {
    //cout << this_thread::get_id() << "begin" << endl;
    LOG_INFO << a << " " << b << c;
    return ;
}


int main() {
    // SET_LEVEL(ttw::LogLevel::DEBUG);
    LOG_INFO << "hellow world" << " " << 123 ;
    LOG_WARNING << "hellow world" << " " << 123;
    LOG_DEBUG << "hellow world" << " " << 123;
    LOG_ERROE << "hellow world" << " " << 123;
    LOG_FATAL << "hellow world" << " " << 123;
    
    ttw::ThreadPool tp;
    for(int i = 0; i < 100; i++) {
        tp.add_task(func, i, 2 * i, 3 * i);

    }
    tp.stop();
   
    return 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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
/*************************************************************************
        > File Name: 1.thredd_pool.cpp
        > Author: 秃头王
        > Mail: 1658339000@qq.com
        > Created Time: 2022年07月25日 星期一 18时08分16秒
 ************************************************************************/

#include <iostream>
#include <thread>
using namespace std;

#define BEGINS(x) namespace x {
#define ENDS(x) }

BEGINS(thread_usage)

void func() {
    cout << "hello world" << endl;
}

void print(int a, int b) {
    cout << a << " " << b << endl;
    return ;
}

int main() {
    // 线程
    thread t1(func);
    t1.join();
    thread t2(print, 1, 2);
    t2.join();
    return 0;
}

ENDS(thread_usage)

BEGINS(prime_count_test)

bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

int prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}

// 给多线程执行的入口函数
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin" << endl;
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "done" << endl;

}

int main() {
    #define batch 5000000
    thread *t[10];
    int ret[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(worker, j, j + batch - 1, ref(ret[i]));
    }

    for (auto x : t) x->join();
    int ans = 0;
    for(auto x : ret) ans += x;
    cout << ans << endl;
    for(auto x : t) delete x;
    #undef batch
    return 0;
}

ENDS(prime_count_test)

#include <mutex>

BEGINS(prime_count_test1)

int ans = 0;
std::mutex m_mutex;


bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

void prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        // 加锁的地方就是临界区
        // 抢占胡互斥锁, 如果抢占上了的话才对ans + 1, 出了作用域就会呗释放
        unique_lock<mutex> lock(m_mutex);
        ans += is_prime(i);
        lock.unlock();
        // 对这种加操作的时候,有可能会发生多线程抢占资源的时候的时候导致数据不正确
        // 临界资源
    }
    return ;
}


int main() {
    #define batch 5000000
    thread *t[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(prime_count, j, j + batch - 1);
    }

    for (auto x : t) x->join();
    for(auto x : t) delete x;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(prime_count_test1)

// 第二种加锁
BEGINS(prime_count_test2)

int ans = 0;
std::mutex m_mutex;


bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

void prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        int ret = is_prime(i);
        __sync_fetch_and_add(&ans, ret);
    }
    return ;
}


int main() {
    #define batch 5000000
    thread *t[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(prime_count, j, j + batch - 1);
    }

    for (auto x : t) x->join();
    for(auto x : t) delete x;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(prime_count_test1)

#include <functional>
BEGINS(tast_test)

// 函数方法绑定到一起的
class Task {
public :

    template<typename FUN_T, typename ...ARGS>
    Task(FUN_T func, ARGS... args) {
        // 用forward向下传时候可以保证准确的值传递
        this->func = bind(func, forward<ARGS>(args)...);
        // 右值传递时候可能就会出错
        // this->func = bind(func, fargs...);
    }

    void run() {
        func();
        return ;
    }

private :
    function<void()> func;
};

void test() {
    cout << "hello world : function test" << endl;
    return ;
}

void func(int a, int b) {
    cout << "function " << a << " " << b << endl;
    return ;
}

void add_one(int &n) {
    n += 1;
    return ;
}

int main() {
    Task t(func, 3, 4);
    t.run();
    Task t2(test);
    t2.run();
    int n = 1;
    cout << "n = " << n << endl;
    Task t3(add_one, n);
    // 加上 ref() 为引用是因为在模板在参数推导时候会帮你推到成引用
    Task t4(add_one, ref(n));

    t4.run();

    t3.run();
    t3.run();
    t3.run();
    t3.run();
    t3.run();
    cout << "n = " << n << endl;
    return 0;
}

ENDS(task_test)

#include <vector>
#include <unordered_map>
#include <queue>
#include <condition_variable>

BEGINS(thread_pool_test)

// 函数方法绑定到一起的
class Task {
public :

    template<typename FUN_T, typename ...ARGS>
    Task(FUN_T func, ARGS... args) {
        // 用forward向下传时候可以保证准确的值传递
        this->func = bind(func, forward<ARGS>(args)...);
        // 右值传递时候可能就会出错
        // this->func = bind(func, fargs...);
    }

    void run() {
        func();
        return ;
    }

private :
    function<void()> func;
};


// 线程池类
class ThreadPool {
public :
    ThreadPool(int n = 1) : thread_size(n), threads(n), starting(false) {
        this->start();
    }

    // 取任务、执行任务
    void worker() {
        auto id = this_thread::get_id();
        running[id] = true;
        while(running[id]) {
            Task *t = get_task();
            t->run();
            delete t;
        }
    }

    void start() {
        if(starting == true) return ;
        for(int i = 0; i < thread_size; i++) {
            threads[i] = new thread(&ThreadPool::worker, this);
        }
        starting = true;
        return ;
    }

    void stop() {
        if(starting == false) return ;
        for(int i = 0; i < threads.size(); i++) {
            this->add_task(&ThreadPool::stop_running, this);
        }
        for(int i = 0; i < threads.size(); i++) {
            threads[i]->join();
        }
        for(int i = 0; i < threads.size(); i++) {
            delete threads[i];
            threads[i] = nullptr;
        }
        starting = false;
        return ;
    }

    template<typename FUNC_T, typename ...ARGS>
    void add_task(FUNC_T func, ARGS... args) {
        unique_lock<mutex> lock(m_mutex);
        tasks.push(new Task(func, forward<ARGS>(args)...));

        // 通知所有等待着任务的线程
        m_cond.notify_one();
        return ;
    }

    ~ThreadPool() {
        this->stop();
        while(!tasks.empty()) {
            delete tasks.front();
            tasks.pop();
        }
        return ;
    }

private :
    Task *get_task() {
       unique_lock<mutex> lock(m_mutex);
        // 防止虚假唤醒 wait 会自动释放锁
        while (tasks.empty()) m_cond.wait(lock);
        Task *t = tasks.front();
        tasks.pop();
        return t;
    }

    void stop_running() {
        auto id = this_thread::get_id();
        running[id] = false;
        return ;
    }

    bool starting;
    int thread_size;
    vector<thread *> threads;
    unordered_map<decltype(this_thread::get_id()), bool> running;
    queue<Task *> tasks;

    std::condition_variable m_cond;
    std::mutex m_mutex;

};

bool is_prime(int x){
    for(int i = 2; i <= x / i; i++) {
        if(x % i == 0) return false;
    }
    return true;
}

int prime_count(int l, int r) {
    int ans = 0;
    for(int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}

// 给多线程执行的入口函数
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin" << endl;
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "done" << endl;

}

int main() {
    ThreadPool tp(5);
     #define batch 5000000
    thread *t[10];
    int ret[10];
    for(int i = 0,j = 1; i < 10; i++, j += batch) {
        tp.add_task(worker, j, j + batch - 1, ref(ret[i]));
    }
    tp.stop();
    int ans = 0;
    for(auto x : ret) ans +=x ;
    cout << ans << endl;
    #undef batch
    return 0;
}

ENDS(thread_pool_test)

int main() {
    // thread_usage::main();
    // prime_count_test::main();
    // prime_count_test1::main();
    // prime_count_test2::main();
    // task_test::main();
    thread_pool_test::main();
    return 0;
}

源码文件*.cpp

点我跳转:https://github.com/qzwl123/C-/tree/main/%E7%BA%BF%E7%A8%8B%E6%B1%A0

博客内容遵循 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 协议

本文永久链接是：https://tutouwang.vercel.app/2022/07/26/OOP/%E7%BA%BF%E7%A8%8B%E6%B1%A0/

更新于：2022年9月22日

评论