多线程-生产者和消费者模式_多线程消费者和生产者模式

1.简单实现多线程

多线程是多任务处理的一种特殊形式，多线程处理允许让一个进程中同时运行两个或两个以上的线程。这样的话，能更加充分发挥计算机的性能，并高效完成用户的任务。

多线程实现的三步骤：

第一步：加入头文件

#include <thread>
1

第二步：子线程的实现

void worker()	  // 子线程的函数代码 
{
	...
}
1
2
3
4

第三步：创建线程

std::thread t(worker);
t.join();//或者t.detach();
1
2

join和detach的对比：
join()函数是一个等待线程完成函数，主线程需要等待子线程运行结束了才可以往下执行
detach()函数是子线程的分离函数，当调用该函数后，线程就被分离到后台运行，主线程不需要等待该线程结束才往下执行

基于join的多线程

#include <stdio.h>
#include <thread>	//多线程的头文件
using namespace std;
void worker()	  // 子线程的函数代码 
{
    printf("hello thread!\n");
    this_thread::sleep_for(chrono::milliseconds(1000));
}
int main(){
    std::thread t(worker);  // 创建子线程
	t.join();				// 等待子线程结束
    printf("done!\n");
    this_thread::sleep_for(chrono::milliseconds(2000));
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

上面代码的运行结果：显示打印hello thread，然后等待1秒后，在打印done。再等待2秒后，程序结束。

基于detach的多线程

#include <stdio.h>
#include <thread>	//多线程的头文件
using namespace std;
void worker()	  // 子线程的函数代码 
{
    printf("hello thread!\n");
    this_thread::sleep_for(chrono::milliseconds(1000));
}
int main(){
    std::thread t(worker);  // 创建子线程
	t.detach();				// 等待子线程结束
    printf("done!\n");
    this_thread::sleep_for(chrono::milliseconds(2000));
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

上面代码的运行结果：显示打印done，然后打印hello thread。再等待2秒后，程序结束。–此时，main函数和子线程worker的执行是分开执行的，主函数不需要等待worker执行结束才往下执行。

在类中使用多线程

class Infer
{
public:
    Infer()
    {
        worker_thread_ = thread(&Infer::infer_worker,this);
    }
    ~Infer()
    {
        if(worker_thread_.joinable())
            worker_thread_.join();
    }
private:
    void infer_worker()
    {

    }
private:
    thread worker_thread_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2.多线程-锁mutex

在c++的多线程编程中，对于内存中的变量可能出现不同线程的同时读取写，这就应该引入锁来提供“访问保护”。即通过锁实现在一个线程对某个变量进行读写操作的时候，其他线程不能访问该变量。

基于lock和unlock的加锁

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>
#include <chrono>
#include <stdexcept>

int counter = 0;
std::mutex mtx; // 保护counter

void increase(int time) {
    for (int i = 0; i < time; i++) {
        mtx.lock();
        // 当前线程休眠1毫秒
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        counter++;
        mtx.unlock();
    }
}

int main(int argc, char** argv) {
    std::thread t1(increase, 10000);
    std::thread t2(increase, 10000);
    t1.join();
    t2.join();
    std::cout << "counter:" << counter << std::endl;
    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

基于std::lock_guard的加锁

使用lock和unlock的加锁，往往容易忘记加unlock的解锁代码。所以，引入了lock_guard

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>
#include <chrono>
#include <stdexcept>

int counter = 0;
std::mutex mtx; // 保护counter

void increase(int time) {
    for (int i = 0; i < time; i++) {
        std::lock_guard<std::mutex> sd(mtx);
        // 当前线程休眠1毫秒
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        counter++;
    }
}

int main(int argc, char** argv) {
    std::thread t1(increase, 100);
    std::thread t2(increase, 100);
    t1.join();
    t2.join();
    std::cout << "counter:" << counter << std::endl;
    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

死锁

多线程容易出现的常见问题：发生死锁。

死锁的例子：线程T1持有锁L1并且申请获得锁L2，而线程T2持有锁L2并且申请获得锁L3，而线程T3持有锁L3并且申请获得锁L1，这样导致了一个锁依赖的环路：T1依赖T2的锁L2，T2依赖T3的锁L3，而T3依赖T1的锁L1。从而导致了死锁。

避免的死锁的方法：
（1）.如果出现两个线程，都是先锁锁A，再锁锁B，就能避免死锁的现象.（即不要一个线程，先锁锁A，再锁锁B；另外一个线程，先锁锁B，再锁锁A；这就很容易出现死锁问题）
（2）.使用std::lock(锁A, 锁B);进行加锁

3.多线程-condition_variable条件变量

condition_variable类是利用线程间共享的变量进行同步的一种机制，用于阻塞一个线程或者同时阻塞多个线程，直到另一个线程修改共享变量并通知condition_variable.

条件变量的使用总是和一个互斥量结合在一起

条件变量的三步骤：

1. 锁（锁住共享变量，线程独占），锁使用的是 std::unique_lock< std::mutex >
2. wait等待（等待通知条件变量，变化的共享变量是否满足条件）
3. 当notify通知条件变量、超时过期或发生虚假唤醒时，线程被唤醒，互斥锁unlock被原子地重新获取。然后，线程应该检查条件，如果唤醒是假的，则继续等待

#include <iostream>
#include <condition_variable>
using namespace std;
mutex wait_mutex;
condition_variable wait_condition_variable;

// 等待线程函数
void wait_thread_func()
{
	unique_lock<mutex> lock(wait_mutex);
	cout << "等待线程(" << this_thread::get_id() << "): 开始等待通知..." << endl;
	wait_condition_variable.wait(lock); //void wait(std::unique_lock<std::mutex>& lock, Predicate pred); //Predicate 谓词函数，可以普通函数或者lambda表达式
	cout << "等待线程(" << this_thread::get_id() << "): 继续执行代码..." << endl;
}

int main()
{
	thread wait_thread(wait_thread_func);

	this_thread::sleep_for(1s); // 等待1秒后进行通知
	cout << "通知线程(" << this_thread::get_id() << "): 开始通知等待线程..." << endl;
	wait_condition_variable.notify_one();
	wait_thread.join();
	cout << "--- main结束 ---" << endl;
}
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

4.多线程-promise

future和promise的作用是在不同线程之间传递数据。使用指针也可以完成数据的传递，但是指针非常危险，因为互斥量不能阻止指针的访问；而且指针的方式传递的数据是固定的，如果更改数据类型，那么还需要更改有关的接口，比价麻烦；promise支持泛型的操作，更加方便编程处理。

#include <iostream>
#include <functional>
#include <future>
#include <thread>
#include <chrono>
#include <cstdlib>

void thread_set_promise(std::promise<int>& promiseObj) {
    std::cout << "In a thread, making data...\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    promiseObj.set_value(35);
    std::cout << "Finished\n";
}

int main() {
    std::promise<int> promiseObj;
    std::future<int> futureObj = promiseObj.get_future();//promise和future绑定，用于获取线程中promiseObj的值
    std::thread t(&thread_set_promise, std::ref(promiseObj));
    std::cout << futureObj.get() << std::endl;// futureObj.get()：等待futureObj被赋值，获取futureObj的值
    t.join();
    system("pause");
    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

5.生产者和消费者模式概述

概念：多个线程进行生产，同时多个线程进行消费，两种角色通过内存缓冲区进行通信。

即所谓生产者消费者问题实际上主要是包含了两类线程：一类是生产者线程用于生产数据；一类是消费者线程用于消费数据。为了解耦生产者和消费者的关系，通常会采用共享的数据区域，就像是一个仓库，生产者生产数据之后直接放置在共享数据区中，并不需要关心消费者的行为；消费者只需要从共享数据区中去获取数据，并不需要关心生产者的行为

优点：极大的解决了代码之间的耦合程度
注释：解耦的意思假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化，可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也就相应降低了。