线程

线程创建

C++11引入了强大的多线程支持，主要通过<thread>头文件提供。以下是详细介绍：

使用函数创建线程


x
1
#include <iostream>
2
#include <thread>
3

4
void hello() {
5
    std::cout << "Hello from thread!" << std::endl;
6
}
7

8
int main() {
9
    std::thread t(hello);  // 创建线程
10
    t.join();              // 等待线程完成
11
    return 0;
12
}

使用Lambda表达式


xxxxxxxxxx
10
1
#include <iostream>
2
#include <thread>
3

4
int main() {
5
    std::thread t([]() {
6
        std::cout << "Lambda thread!" << std::endl;
7
    });
8
    t.join();
9
    return 0;
10
}

使用类成员函数


xxxxxxxxxx
16
1
#include <iostream>
2
#include <thread>
3

4
class MyClass {
5
public:
6
    void member_function() {
7
        std::cout << "Member function thread!" << std::endl;
8
    }
9
};
10

11
int main() {
12
    MyClass obj;
13
    std::thread t(&MyClass::member_function, &obj);
14
    t.join();
15
    return 0;
16
}

线程管理

join() 和 detach()


xxxxxxxxxx
19
1
#include <iostream>
2
#include <thread>
3
#include <chrono>
4

5
void worker() {
6
    std::this_thread::sleep_for(std::chrono::seconds(2));
7
    std::cout << "Worker finished" << std::endl;
8
}
9

10
int main() {
11
    std::thread t(worker);
12
    
13
    // 选择一种方式：
14
    t.join();    // 等待线程完成
15
    // 或者
16
    // t.detach(); // 分离线程，让它独立运行
17
    
18
    return 0;
19
}

检查线程状态


xxxxxxxxxx
15
1
#include <iostream>
2
#include <thread>
3

4
int main() {
5
    std::thread t([]() {
6
        std::cout << "Thread running" << std::endl;
7
    });
8
    
9
    if (t.joinable()) {
10
        std::cout << "Thread is joinable" << std::endl;
11
        t.join();
12
    }
13
    
14
    return 0;
15
}

线程参数传递

按值传递


xxxxxxxxxx
16
1
#include <iostream>
2
#include <thread>
3

4
void print_value(int x, std::string str) {
5
    std::cout << "Value: " << x << ", String: " << str << std::endl;
6
}
7

8
int main() {
9
    int num = 42;
10
    std::string text = "Hello";
11
    
12
    std::thread t(print_value, num, text);
13
    t.join();
14
    
15
    return 0;
16
}

按引用传递


xxxxxxxxxx
18
1
#include <iostream>
2
#include <thread>
3

4
void modify_value(int& x) {
5
    x = 100;
6
}
7

8
int main() {
9
    int num = 42;
10
    
11
    // 使用std::ref传递引用
12
    std::thread t(modify_value, std::ref(num));
13
    t.join();
14
    
15
    std::cout << "Modified value: " << num << std::endl; // 输出100
16
    
17
    return 0;
18
}

锁

基础锁类型

互斥锁

std::mutex是最基本的互斥锁，同一时间只允许一个线程访问。


xxxxxxxxxx
25
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4

5
std::mutex mtx;
6
int counter = 0;
7

8
void increment() {
9
    for (int i = 0; i < 1000; ++i) {
10
        mtx.lock();    // 加锁
11
        ++counter;
12
        mtx.unlock();  // 解锁
13
    }
14
}
15

16
int main() {
17
    std::thread t1(increment);
18
    std::thread t2(increment);
19
    
20
    t1.join();
21
    t2.join();
22
    
23
    std::cout << "Counter: " << counter << std::endl;
24
    return 0;
25
}

递归互斥锁

std::recursive_mutex允许同一线程多次获取同一个锁。


xxxxxxxxxx
23
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4

5
std::recursive_mutex rec_mtx;
6

7
void recursive_function(int depth) {
8
    rec_mtx.lock();
9
    
10
    std::cout << "Depth: " << depth << std::endl;
11
    
12
    if (depth > 0) {
13
        recursive_function(depth - 1);  // 同一线程再次获取锁
14
    }
15
    
16
    rec_mtx.unlock();
17
}
18

19
int main() {
20
    std::thread t(recursive_function, 3);
21
    t.join();
22
    return 0;
23
}

定时互斥锁

std::timed_mutex是支持超时的互斥锁。


xxxxxxxxxx
27
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4
#include <chrono>
5

6
std::timed_mutex timed_mtx;
7

8
void worker(int id) {
9
    // 尝试在2秒内获取锁
10
    if (timed_mtx.try_lock_for(std::chrono::seconds(2))) {
11
        std::cout << "Thread " << id << " got the lock" << std::endl;
12
        std::this_thread::sleep_for(std::chrono::seconds(3));
13
        timed_mtx.unlock();
14
    } else {
15
        std::cout << "Thread " << id << " timeout" << std::endl;
16
    }
17
}
18

19
int main() {
20
    std::thread t1(worker, 1);
21
    std::thread t2(worker, 2);
22
    
23
    t1.join();
24
    t2.join();
25
    
26
    return 0;
27
}

递归定时互斥锁

std::recursive_timed_mutex是结合了递归和定时功能的互斥锁。


xxxxxxxxxx
13
1
std::recursive_timed_mutex rec_timed_mtx;
2

3
void timed_recursive_function(int depth) {
4
    if (rec_timed_mtx.try_lock_for(std::chrono::milliseconds(100))) {
5
        std::cout << "Depth: " << depth << std::endl;
6
        
7
        if (depth > 0) {
8
            timed_recursive_function(depth - 1);
9
        }
10
        
11
        rec_timed_mtx.unlock();
12
    }
13
}

锁管理器

lock_guard

最简单的RAII锁管理器，构造时加锁，析构时自动解锁。


xxxxxxxxxx
36
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4

5
std::mutex mtx;
6
int shared_data = 0;
7

8
void safe_increment() {
9
    std::lock_guard<std::mutex> lock(mtx);  // 自动加锁
10
    ++shared_data;
11
    // 函数结束时自动解锁
12
}
13

14
void risky_function() {
15
    std::lock_guard<std::mutex> lock(mtx);
16
    ++shared_data;
17
    
18
    if (shared_data > 5) {
19
        throw std::runtime_error("Error!");  // 异常时也会自动解锁
20
    }
21
}
22

23
int main() {
24
    std::vector<std::thread> threads;
25
    
26
    for (int i = 0; i < 10; ++i) {
27
        threads.emplace_back(safe_increment);
28
    }
29
    
30
    for (auto& t : threads) {
31
        t.join();
32
    }
33
    
34
    std::cout << "Shared data: " << shared_data << std::endl;
35
    return 0;
36
}

unique_lock

更灵活的锁管理器，支持延迟加锁、条件变量等。


xxxxxxxxxx
47
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4
#include <condition_variable>
5

6
std::mutex mtx;
7
std::condition_variable cv;
8
bool ready = false;
9

10
void worker() {
11
    std::unique_lock<std::mutex> lock(mtx);
12
    
13
    // 等待条件满足
14
    cv.wait(lock, []{ return ready; });
15
    
16
    std::cout << "Worker thread proceeding..." << std::endl;
17
}
18

19
void setter() {
20
    std::this_thread::sleep_for(std::chrono::seconds(1));
21
    
22
    {
23
        std::unique_lock<std::mutex> lock(mtx);
24
        ready = true;
25
    }
26
    
27
    cv.notify_one();
28
}
29

30
// 手动控制锁的示例
31
void flexible_locking() {
32
    std::unique_lock<std::mutex> lock(mtx, std::defer_lock);  // 延迟加锁
33
    
34
    // 做一些不需要锁的工作
35
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
36
    
37
    lock.lock();  // 手动加锁
38
    // 临界区代码
39
    lock.unlock();  // 手动解锁
40
    
41
    // 做更多不需要锁的工作
42
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
43
    
44
    lock.lock();  // 再次加锁
45
    // 更多临界区代码
46
    // 析构时自动解锁
47
}

shared_lock

用于读写锁的共享锁定。


xxxxxxxxxx
20
1
#include <iostream>
2
#include <thread>
3
#include <shared_mutex>  // C++14
4
#include <vector>
5

6
std::shared_mutex rw_mutex;
7
int shared_data = 0;
8

9
void reader(int id) {
10
    std::shared_lock<std::shared_mutex> lock(rw_mutex);  // 共享锁
11
    std::cout << "Reader " << id << " reads: " << shared_data << std::endl;
12
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
13
}
14

15
void writer(int id) {
16
    std::unique_lock<std::shared_mutex> lock(rw_mutex);  // 独占锁
17
    ++shared_data;
18
    std::cout << "Writer " << id << " writes: " << shared_data << std::endl;
19
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
20
}

避免死锁

同时锁定多个互斥锁，避免死锁。


xxxxxxxxxx
33
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4

5
std::mutex mtx1, mtx2;
6

7
void worker1() {
8
    std::lock(mtx1, mtx2);  // 同时锁定两个互斥锁
9
    std::lock_guard<std::mutex> lock1(mtx1, std::adopt_lock);
10
    std::lock_guard<std::mutex> lock2(mtx2, std::adopt_lock);
11
    
12
    std::cout << "Worker 1 acquired both locks" << std::endl;
13
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
14
}
15

16
void worker2() {
17
    std::lock(mtx2, mtx1);  // 不同的顺序，但不会死锁
18
    std::lock_guard<std::mutex> lock1(mtx2, std::adopt_lock);
19
    std::lock_guard<std::mutex> lock2(mtx1, std::adopt_lock);
20
    
21
    std::cout << "Worker 2 acquired both locks" << std::endl;
22
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
23
}
24

25
int main() {
26
    std::thread t1(worker1);
27
    std::thread t2(worker2);
28
    
29
    t1.join();
30
    t2.join();
31
    
32
    return 0;
33
}


xxxxxxxxxx
27
1
// 策略1：固定顺序获取锁
2
void fixed_order_locking() {
3
    // 总是按照地址顺序获取锁
4
    std::mutex* first = &mtx1 < &mtx2 ? &mtx1 : &mtx2;
5
    std::mutex* second = &mtx1 < &mtx2 ? &mtx2 : &mtx1;
6
    
7
    std::lock_guard<std::mutex> lock1(*first);
8
    std::lock_guard<std::mutex> lock2(*second);
9
}
10

11
// 策略2：使用std::lock
12
void simultaneous_locking() {
13
    std::lock(mtx1, mtx2);
14
    std::lock_guard<std::mutex> lock1(mtx1, std::adopt_lock);
15
    std::lock_guard<std::mutex> lock2(mtx2, std::adopt_lock);
16
}
17

18
// 策略3：使用超时
19
void timeout_locking() {
20
    std::unique_lock<std::timed_mutex> lock1(timed_mtx);
21
    
22
    if (auto lock2 = std::unique_lock<std::timed_mutex>(timed_mtx, std::try_to_lock)) {
23
        // 成功获取两个锁
24
    } else {
25
        // 处理获取锁失败的情况
26
    }
27
}

try_lock系列

非阻塞的锁获取尝试。


xxxxxxxxxx
32
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4

5
std::mutex mtx;
6

7
void try_lock_example() {
8
    if (mtx.try_lock()) {
9
        std::cout << "Got the lock!" << std::endl;
10
        std::this_thread::sleep_for(std::chrono::seconds(1));
11
        mtx.unlock();
12
    } else {
13
        std::cout << "Failed to get the lock" << std::endl;
14
    }
15
}
16

17
void try_lock_multiple() {
18
    std::mutex mtx1, mtx2, mtx3;
19
    
20
    // 尝试同时获取多个锁
21
    int result = std::try_lock(mtx1, mtx2, mtx3);
22
    
23
    if (result == -1) {
24
        // 成功获取所有锁
25
        std::cout << "Got all locks" << std::endl;
26
        mtx1.unlock();
27
        mtx2.unlock();
28
        mtx3.unlock();
29
    } else {
30
        std::cout << "Failed at lock " << result << std::endl;
31
    }
32
}

锁的构造选项

defer_lock

延迟加锁，构造时不立即获取锁。


xxxxxxxxxx
11
1
std::mutex mtx;
2

3
void deferred_locking() {
4
    std::unique_lock<std::mutex> lock(mtx, std::defer_lock);
5
    
6
    // 做一些不需要锁的工作
7
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
8
    
9
    lock.lock();  // 现在才获取锁
10
    // 临界区代码
11
}

try_to_lock

尝试获取锁，不阻塞。


xxxxxxxxxx
10
1
void try_to_lock_example() {
2
    std::unique_lock<std::mutex> lock(mtx, std::try_to_lock);
3
    
4
    if (lock.owns_lock()) {
5
        std::cout << "Successfully acquired lock" << std::endl;
6
        // 临界区代码
7
    } else {
8
        std::cout << "Failed to acquire lock" << std::endl;
9
    }
10
}

adopt_lock

接管已经获取的锁。


xxxxxxxxxx
8
1
void adopt_lock_example() {
2
    mtx.lock();  // 手动获取锁
3
    
4
    std::lock_guard<std::mutex> lock(mtx, std::adopt_lock);  // 接管锁
5
    
6
    // 临界区代码
7
    // lock析构时会自动释放锁
8
}

锁的性能对比

锁类型	性能	特性	使用场景
mutex	高	基础互斥	一般同步
recursive_mutex	中	递归锁定	递归函数
timed_mutex	中	超时机制	需要超时控制
shared_mutex	中	读写分离	读多写少

条件变量

C++中使用条件变量可以使用std::condition_variable。


xxxxxxxxxx
50
1
#include <iostream>
2
#include <thread>
3
#include <mutex>
4
#include <condition_variable>
5
#include <queue>
6

7
std::mutex mtx;
8
std::condition_variable cv;
9
std::queue<int> data_queue;
10
bool finished = false;
11

12
void producer() {
13
    for (int i = 0; i < 10; ++i) {
14
        std::unique_lock<std::mutex> lock(mtx);
15
        data_queue.push(i);
16
        std::cout << "Produced: " << i << std::endl;
17
        cv.notify_one(); // 通知等待的线程
18
    }
19
    
20
    std::unique_lock<std::mutex> lock(mtx);
21
    finished = true;
22
    cv.notify_all();
23
}
24

25
void consumer() {
26
    while (true) {
27
        std::unique_lock<std::mutex> lock(mtx);
28
        
29
        // 等待条件满足
30
        cv.wait(lock, []() { return !data_queue.empty() || finished; });
31
        
32
        while (!data_queue.empty()) {
33
            int value = data_queue.front();
34
            data_queue.pop();
35
            std::cout << "Consumed: " << value << std::endl;
36
        }
37
        
38
        if (finished) break;
39
    }
40
}
41

42
int main() {
43
    std::thread prod(producer);
44
    std::thread cons(consumer);
45
    
46
    prod.join();
47
    cons.join();
48
    
49
    return 0;
50
}

Atomic类型

实现原理：

CAS机制：会将变量的预期值A与在内存中的值V进行比较，如果V与A是相等，就可以将该值改为B；如果V与A的值是不相等，那么就不能将其改为新值B。


xxxxxxxxxx
23
1
#include <iostream>
2
#include <thread>
3
#include <atomic>
4

5
std::atomic<int> atomic_counter(0);
6

7
void atomic_increment() {
8
    for (int i = 0; i < 1000; ++i) {
9
        atomic_counter++; // 原子操作，无需显式锁
10
    }
11
}
12

13
int main() {
14
    std::thread t1(atomic_increment);
15
    std::thread t2(atomic_increment);
16
    
17
    t1.join();
18
    t2.join();
19
    
20
    std::cout << "Atomic counter: " << atomic_counter << std::endl;
21
    
22
    return 0;
23
}

生产者与消费者模型

类图

代码


xxxxxxxxxx
87
1
#include <condition_variable>
2
#include <functional>
3
#include <iostream>
4
#include <mutex>
5
#include <queue>
6
#include <thread>
7

8
using std::condition_variable;
9
using std::cout;
10
using std::endl;
11
using std::mutex;
12
using std::queue;
13
using std::thread;
14
using std::unique_lock;
15

16
class TaskQueue {
17
public:
18
    TaskQueue(size_t capacity) : _capacity(capacity) {}
19
    bool isFull() {
20
        return _queue.size() >= _capacity;
21
    }
22
    bool isEmpty() {
23
        return _queue.empty();
24
    }
25
    void push(const int &value) {
26
        unique_lock<mutex> ul{_mutex};
27
        while (isFull()) {
28
            _notFull.wait(ul);
29
        }
30
        _queue.push(value);
31
        cout << "生产" << value << endl
32
             << "--------" << endl;
33
        _notEmpty.notify_one();
34
    }
35
    int pop() {
36
        unique_lock<mutex> ul{_mutex};
37
        while (isEmpty()) {
38
            _notEmpty.wait(ul);
39
        }
40
        int temp = _queue.front();
41
        _queue.pop();
42
        cout << "消费" << temp << endl
43
             << "--------" << endl;
44
        _notFull.notify_one();
45
        return temp;
46
    }
47

48
private:
49
    size_t _capacity;
50
    queue<int> _queue;
51
    mutex _mutex;
52
    condition_variable _notEmpty;
53
    condition_variable _notFull;
54
};
55

56
class Producer {
57
public:
58
    void produce(TaskQueue &taskQueue) {
59
        srand(clock());
60
        int i = 10;
61
        while (i--) {
62
            taskQueue.push(rand() % 100);
63
        }
64
    }
65
};
66

67
class Consumer {
68
public:
69
    void consume(TaskQueue &taskQueue) {
70
        int i = 10;
71
        while (i--) {
72
            taskQueue.pop();
73
        }
74
    }
75
};
76

77
int main(int argc, char *argv[]) {
78
    TaskQueue taskQueue{5};
79
    Producer producer;
80
    Consumer consumer;
81
    thread producerThread{&Producer::produce, &producer, std::ref(taskQueue)};
82
    thread consumerThread{&Consumer::consume, &consumer, std::ref(taskQueue)};
83

84
    producerThread.join();
85
    consumerThread.join();
86
    return 0;
87
}

说明

TaskQueue（任务队列）

私有成员：
- _capacity: 队列容量
- _queue: 存储整数的队列
- _mutex: 互斥锁
- _notEmpty, _notFull: 条件变量
公有方法：
- 构造函数、判空判满、入队出队操作

Producer（生产者）

包含 produce() 方法，向队列生产数据

Consumer（消费者）

包含 consume() 方法，从队列消费数据

面向对象的线程池

类图

代码


xxxxxxxxxx
147
1
#include <condition_variable>
2
#include <functional>
3
#include <iostream>
4
#include <memory>
5
#include <mutex>
6
#include <queue>
7
#include <thread>
8
#include <vector>
9

10
using std::condition_variable;
11
using std::cout;
12
using std::endl;
13
using std::mutex;
14
using std::queue;
15
using std::thread;
16
using std::unique_lock;
17
using std::unique_ptr;
18
using std::vector;
19

20
class Task {
21
public:
22
    virtual void process() = 0;
23
    virtual ~Task() {}
24
};
25

26
class TaskQueue {
27
public:
28
    TaskQueue(size_t capacity) : _capacity(capacity) {}
29
    bool isFull() {
30
        return _queue.size() >= _capacity;
31
    }
32
    bool isEmpty() {
33
        return _queue.empty();
34
    }
35
    void push(Task *task) {
36
        unique_lock<mutex> ul{_mutex};
37
        while (isFull()) {
38
            _notFull.wait(ul);
39
        }
40
        _queue.push(task);
41
        _notEmpty.notify_one();
42
    }
43
    Task *pop() {
44
        unique_lock<mutex> ul{_mutex};
45
        while (isEmpty() && _flag) {
46
            _notEmpty.wait(ul);
47
        }
48
        if (_flag) {
49
            Task *temp = _queue.front();
50
            _queue.pop();
51
            _notFull.notify_one();
52
            return temp;
53
        }
54
        return nullptr;
55
    }
56
    void wakeAll() {
57
        _flag = false;
58
        _notEmpty.notify_all();
59
    }
60

61
private:
62
    size_t _capacity;
63
    queue<Task *> _queue;
64
    mutex _mutex;
65
    condition_variable _notEmpty;
66
    condition_variable _notFull;
67
    bool _flag = true;
68
};
69

70
class ThreadPool {
71
public:
72
    ThreadPool(size_t threadNum, size_t queueSize)
73
        : _threadNum(threadNum),
74
          _taskQueue(queueSize),
75
          _queueSize(queueSize) {}
76
    void start() {
77
        for (int i = 0; i < _threadNum; ++i) {
78
            _threads.push_back(thread{&ThreadPool::doTask, this});
79
        }
80
    }
81
    void stop() {
82
        while (!_taskQueue.isEmpty()) {
83
            std::this_thread::sleep_for(std::chrono::seconds(1));
84
        }
85
        _isAlive = false;
86
        _taskQueue.wakeAll();
87
        for (auto &th : _threads) {
88
            th.join();
89
        }
90
    }
91
    void addTask(Task *task) {
92
        if (task) {
93
            _taskQueue.push(task);
94
        }
95
    }
96
    Task *getTask() {
97
        return _taskQueue.pop();
98
    }
99
    void doTask() {
100
        while (_isAlive) {
101
            Task *task = getTask();
102
            if (task && _isAlive) {
103
                task->process();
104
            }
105
        }
106
    }
107

108
private:
109
    TaskQueue _taskQueue;
110
    size_t _threadNum;
111
    vector<thread> _threads;
112
    size_t _queueSize;
113
    bool _isAlive = true;
114
};
115

116
int taskNum = 1;
117
mutex mtx;
118

119
class Task1 : public Task {
120
public:
121
    void process() override {
122
        unique_lock<mutex> ul{mtx};
123
        cout << "任务" << taskNum++ << endl;
124
    }
125
};
126

127
class Task2 : public Task {
128
public:
129
    void process() override {
130
        // 任务2
131
    }
132
};
133

134
int main(int argc, char *argv[]) {
135
    ThreadPool pool{3, 10};
136
    pool.start();
137
    vector<unique_ptr<Task>> tasks;
138
    int i = 20;
139
    while (i--) {
140
        unique_ptr<Task> task{new Task1};
141
        pool.addTask(task.get());
142
        tasks.push_back(std::move(task));
143
        cout << i << endl;
144
    }
145
    pool.stop();
146
    return 0;
147
}

说明

1. Task（抽象基类）

类型 : 抽象基类
方法 :
- process() : 纯虚函数，定义任务处理接口
- ~Task() : 虚析构函数，确保正确的多态析构

2. TaskQueue（任务队列）

私有成员 :
- _capacity : 队列容量限制
- _queue : 存储Task指针的队列
- _mutex : 互斥锁，保护共享资源
- _notEmpty : 条件变量，通知队列非空
- _notFull : 条件变量，通知队列未满
- _flag : 布尔标志，控制队列状态
公有方法 :
- TaskQueue(size_t capacity) : 构造函数
- isFull() : 检查队列是否已满
- isEmpty() : 检查队列是否为空
- push(Task *task) : 添加任务（阻塞式）
- pop() : 获取任务（阻塞式）
- wakeAll() : 唤醒所有等待线程

3. ThreadPool（线程池）

私有成员 :
- _taskQueue : 任务队列实例
- _threadNum : 线程数量
- _threads : 工作线程容器
- _queueSize : 队列大小
- _isAlive : 线程池状态标志
公有方法 :
- ThreadPool(size_t threadNum, size_t queueSize) : 构造函数
- start() : 启动线程池
- stop() : 停止线程池
- addTask(Task *task) : 添加任务
- getTask() : 获取任务
- doTask() : 工作线程执行函数

4. Task1和Task2（具体任务实现）

Task1 : 实现了 process() 方法，输出任务编号
Task2 : 实现了 process() 方法，当前为空实现

基于对象的线程池

类图

代码


xxxxxxxxxx
135
1
#include <chrono>
2
#include <condition_variable>
3
#include <cstdlib>
4
#include <ctime>
5
#include <functional>
6
#include <iostream>
7
#include <memory>
8
#include <mutex>
9
#include <queue>
10
#include <thread>
11
#include <vector>
12

13
using std::condition_variable;
14
using std::cout;
15
using std::endl;
16
using std::mutex;
17
using std::queue;
18
using std::thread;
19
using std::unique_lock;
20
using std::unique_ptr;
21
using std::vector;
22

23
using Task = std::function<void()>;
24

25
class MyTask {
26
public:
27
    void process(int x) {
28
        // 参数无作用，通过bind绑定改变调用形式
29
        srand(clock());
30
        int num = rand() % 100;
31
        cout << "MyTask::process(): " << num << endl;
32
    }
33
};
34

35
class TaskQueue {
36
public:
37
    TaskQueue(size_t capacity) : _capacity(capacity) {}
38
    bool isFull() {
39
        return _queue.size() >= _capacity;
40
    }
41
    bool isEmpty() {
42
        return _queue.empty();
43
    }
44
    void push(Task &&task) {
45
        unique_lock<mutex> ul{_mutex};
46
        while (isFull()) {
47
            _notFull.wait(ul);
48
        }
49
        _queue.push(std::move(task));
50
        _notEmpty.notify_one();
51
    }
52
    Task pop() {
53
        unique_lock<mutex> ul{_mutex};
54
        while (isEmpty() && _flag) {
55
            _notEmpty.wait(ul);
56
        }
57
        if (_flag && !_queue.empty()) {
58
            Task temp = std::move(_queue.front());
59
            _queue.pop();
60
            _notFull.notify_one();
61
            return temp;
62
        }
63
        return Task{};
64
    }
65
    void wakeAll() {
66
        _flag = false;
67
        _notEmpty.notify_all();
68
    }
69

70
private:
71
    size_t _capacity;
72
    queue<Task> _queue;
73
    mutex _mutex;
74
    condition_variable _notEmpty;
75
    condition_variable _notFull;
76
    bool _flag = true;
77
};
78

79
class ThreadPool {
80
public:
81
    ThreadPool(size_t threadNum, size_t queueSize)
82
        : _threadNum(threadNum),
83
          _taskQueue(queueSize),
84
          _queueSize(queueSize) {}
85
    void start() {
86
        for (int i = 0; i < _threadNum; ++i) {
87
            _threads.push_back(thread{&ThreadPool::doTask, this});
88
        }
89
    }
90
    void stop() {
91
        while (!_taskQueue.isEmpty()) {
92
            std::this_thread::sleep_for(std::chrono::seconds(1));
93
        }
94
        _isAlive = false;
95
        _taskQueue.wakeAll();
96
        for (auto &th : _threads) {
97
            th.join();
98
        }
99
    }
100
    void addTask(Task &&task) {
101
        if (task) {
102
            _taskQueue.push(std::move(task));
103
        }
104
    }
105
    Task getTask() {
106
        return _taskQueue.pop();
107
    }
108
    void doTask() {
109
        while (_isAlive) {
110
            Task task = getTask();
111
            if (task && _isAlive) {
112
                task();
113
            }
114
        }
115
    }
116

117
private:
118
    TaskQueue _taskQueue;
119
    size_t _threadNum;
120
    vector<thread> _threads;
121
    size_t _queueSize;
122
    bool _isAlive = true;
123
};
124

125
int main(int argc, char *argv[]) {
126
    unique_ptr<MyTask> task(new MyTask);
127
    ThreadPool pool{3, 10};
128
    pool.start();
129
    int count = 20;
130
    while (count--) {
131
        pool.addTask(std::bind(&MyTask::process, task.get(), 1));
132
    }
133
    pool.stop();
134
    return 0;
135
}

说明

类

1. Task 类型别名


xxxxxxxxxx
1
1
using Task = std::function<void()>;

作用：定义任务的统一接口
特点：可以封装任何无参数、无返回值的可调用对象
支持：函数指针、lambda 表达式、std::bind 绑定的函数对象等

2. MyTask 类

功能：具体的任务实现类，演示如何创建可执行的任务。

核心方法：

process(int x)：任务的具体执行逻辑
- 生成随机数并输出
- 参数 x 在此实现中无实际作用，主要用于演示 std::bind 的使用

3. TaskQueue 类

功能：线程安全的任务队列，实现生产者-消费者模式。

私有成员

_capacity：队列最大容量
_queue：存储任务的队列容器
_mutex：保护队列的互斥锁
_notEmpty：队列非空条件变量
_notFull：队列未满条件变量
_flag：控制队列状态的标志位

公有方法

构造函数


xxxxxxxxxx
1
1
TaskQueue(size_t capacity)

初始化指定容量的任务队列

状态查询


xxxxxxxxxx
2
1
bool isFull()    // 检查队列是否已满
2
bool isEmpty()   // 检查队列是否为空

任务操作


xxxxxxxxxx
3
1
void push(Task &&task)  // 添加任务到队列
2
Task pop()              // 从队列取出任务
3
void wakeAll()          // 唤醒所有等待线程并停止队列

线程安全机制

互斥访问：使用 std::mutex 保护队列操作
条件等待：
- 队列满时，生产者线程等待 _notFull 条件
- 队列空时，消费者线程等待 _notEmpty 条件
优雅退出：通过 _flag 标志位控制线程退出

4. ThreadPool 类

功能：线程池管理器，负责线程生命周期和任务分发。

私有成员

_taskQueue：任务队列实例
_threadNum：线程池中的线程数量
_threads：存储工作线程的容器
_queueSize：任务队列大小
_isAlive：线程池运行状态标志

公有方法

构造函数


xxxxxxxxxx
1
1
ThreadPool(size_t threadNum, size_t queueSize)

threadNum：工作线程数量
queueSize：任务队列容量

生命周期管理


xxxxxxxxxx
2
1
void start()  // 启动线程池
2
void stop()   // 停止线程池

任务管理


xxxxxxxxxx
3
1
void addTask(Task &&task)  // 添加任务到线程池
2
Task getTask()             // 获取待执行任务
3
void doTask()              // 工作线程的主循环

工作流程

1. 初始化阶段

创建 ThreadPool 实例，指定线程数和队列大小
调用 start() 方法启动指定数量的工作线程
每个工作线程执行 doTask() 方法，进入任务等待循环

2. 任务执行阶段

主线程通过 addTask() 添加任务到队列
工作线程从队列中获取任务并执行
使用条件变量实现高效的线程同步

3. 关闭阶段

等待队列中的所有任务执行完成
设置 _isAlive = false 停止接收新任务
调用 wakeAll() 唤醒所有等待的工作线程
等待所有工作线程结束（join()）