发布于 

C++ 多线程编程

C++ 多线程编程入门实践

创建线程

函数指针

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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;

void show(int id, int count, const string &str);

int main() {
    // 创建线程, 传递参数
    thread t1(show, 1, 10, "hello");
    thread t2(show, 2, 20, "world");
    // 主线程阻塞，等待子线程运行结束
    t1.join();
    t2.join();
    cout << "Bye!" << endl;
    return 0;
}

void show(int id, int count, const string &str) {
    for (int i = 0; i < count; ++i) {
        cout << "Thread " << id << ": " << str << endl;
        sleep(1);
    }
}

编译运行：g++ lesson1.cpp -pthread && ./a.out

函数对象

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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;

class Show {
    private:
        int id;
        int count;
        string str;
    public:
        Show(int i, int c, string s) : id(i), count(c), str(s) {};
        void operator()() const {
            for (int i = 0; i < count; ++i) {
                cout << "Thread " << id << ": " << str << endl;
                sleep(1);
            }
        }
};

int main() {
    // Show s1(1, 10, "Hello");
    // Show s2(2, 10, "World");
    // 类的对象当作函数来用
    // thread t1(s1);
    // thread t2(s2);
    // 构造函数返回匿名对象
    thread t1(Show(1, 10, "Hello"));
    thread t2(Show(2, 10, "World"));
    t1.join();
    t2.join();
    cout << "Bye!" << endl;
    return 0;
}

Lambda表达式

1
2
3
4
5
thread t3([](int id, int count, string str) {
    for (int i = 0;i < count; ++i) {
        cout << "Thread " << id << ": " << str << endl;
    }
}, 3, 10, "good");
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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;

class Show {
    private:
        int id;
        int count;
        string str;
    public:
        Show(int i, int c, string s) : id(i), count(c), str(s) {};
        void operator()() const {
            for (int i = 0; i < count; ++i) {
                cout << "Thread " << id << ": " << str << endl;
                sleep(1);
            }
        }
        void display() {
            for (int i = 0; i <count; ++i) {
                cout << "Thread " << id << ": " << str << endl;
                sleep(1);
            }
        }
};

int main() {
    Show s1(1, 10, "Hello");
    // Show s2(2, 10, "World");
    // 类的对象当作函数来用
    thread t1(s1);
    // thread t2(s2);
    // 构造函数返回匿名对象
    // thread t1( Show(1, 10, "Hello"));
    thread t2( Show(2, 10, "World"));
    // lambda 表达式创建线程
    thread t3([](int id, int count, string str) {
        for (int i = 0;i < count; ++i) {
            cout << "Thread " << id << ": " << str << endl;
        }
    }, 3, 10, "good");
    // 成员函数创建线程
    Show s4(4, 10, "nice");
    thread t4(&Show::display, s4);
    t1.join();
    t2.join();
    t3.join();
    t4.join();
    cout << "Bye!" << endl;
    return 0;
}

detach

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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;


void funct();

class Show {
    private:
        int id;
        int count;
        string str;
    public:
        Show(int i, int c, string s) : id(i), count(c), str(s) {};
        void operator()() const {
            for (int i = 0; i < count; ++i) {
                cout << "Thread " << id << ": " << str << endl;
                sleep(1);
            }
        }
};

int main() {

    funct();

    for (int i = 0; i < 10; ++i) {
        cout << "Main thread is running!" << endl;
        sleep(1);
    }

    cout << "Bye!" << endl;
    return 0;
}

// 在函数创建线程，要注意是否使用到局部变量的引用或者指针
void funct() {
    int id = 1;
    int count = 10;
    string str = "Hello World";
    // 线程访问局部变量的隐患
    // 按值传递没有问题，按引用传递要出问题
    Show s1(id, count, str);
    thread t1(s1);
    // 主线程阻塞
    // t1.join();
    // 线程分离
    t1.detach();
}

thread_local

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
#include <iostream>
#include <thread>
#include <unistd.h>

using namespace std;

void threadFunction(int id);

// 线程之间共享变量 i
int i = 0;
// 线程本地存储
// 每个线程拷贝变量 j, 之间不共享
thread_local int j = 0;

int main() {
    thread t1(threadFunction, 1);
    t1.join();
    thread t2(threadFunction, 2);
    t2.join();
    cout << "i = " << i << ", j = " << j << endl;
}

void threadFunction(int id) {
    cout << "Thread " << id << ": i = " << i << ", j = " << j << endl;
    ++i;
    ++j;
}

move

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
#include <iostream>
#include <thread>
#include <unistd.h>
#include <string>

using namespace std;

void show(int count, const string &str);

int main() {
    thread t1(show, 10, "Hello");
    cout << "Thread1 id = " << t1.get_id() << endl;
    sleep(5);
    // 线程移动 两个id 是一样的
    thread t2 = std::move(t1);
    cout << "Thread2 id = " << t2.get_id() << endl;

    cout << "t1.joinable = " << t1.joinable() << endl;
    cout << "t2.joinable = " << t2.joinable() << endl;
    // 判断子线程是否可以汇合到主线程
    if (t1.joinable()) {
        t1.join();
    }
    if (t2.joinable()) {
        t2.join();
    }
    cout << "t1.joinable = " << t1.joinable() << endl;
    cout << "t2.joinable = " << t2.joinable() << endl;
    // 默认构造的线程
    thread t3;
    cout << "t3.joinable = " << t3.joinable() << endl;
    cout << "Bye!" << endl;
    return 0;
}

void show(int count, const string &str) {
    for (int i = 0; i < count; ++i) {
        cout << str << endl;
        sleep(1);
    }
}

RAII机制

子线程出现异常，导致主线程提前结束，子线程被迫终止。

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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;

void funct();

class Show {
    private:
        int id;
        int count;
        string str;
    public:
        Show(int i, int c, string s) : id(i), count(c), str(s) {};
        void operator()() const {
            for (int i = 0; i < count; ++i) {
                cout << "Thread " << id << ": " << str << endl;
                sleep(1);
            }
        }
};


int main() {

    funct();

    for (int i = 0; i < 10; ++i) {
        cout << "Main thread is running!" << endl;
        sleep(1);
    }

    cout << "Bye!" << endl;
    return 0;
}

void funct() {
    int id = 1;
    int count = 10;
    string str = "Hello World";
    Show s1(id, count, str);
    thread t1(s1);

    int n1, n2;
    cout << "Please enter two numbers: " << endl;
    cin >> n1 >> n2;

    try {
        if (n2 == 0) {
            // 子线程出现异常 导致主线程结束 子线程被迫终止
            throw runtime_error("n2 can't be 0");
        }
        cout << "n1 / n2 = " << n1 / n2 << endl;
    } catch(runtime_error) {
        cout << "Quit with exception" << endl;
        // 主 等 子
        t1.join();
        // 主线程结束
        return;
    }
    t1.join();

}

RAII机制，不需要多次 join()

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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;


void funct();

class Show {
    private:
        int id;
        int count;
        string str;
    public:
        Show(int i, int c, string s) : id(i), count(c), str(s) {};
        void operator()() const {
            for (int i = 0; i < count; ++i) {
                cout << "Thread " << id << ": " << str << endl;
                sleep(1);
            }
        }
};

class thread_guard {
    private:
        thread &g_thread;
    public:
        explicit thread_guard(thread &my_thread) : g_thread(my_thread) {}
        ~thread_guard() {
            if (g_thread.joinable()) {
                g_thread.join();
            }
        }
};

int main() {

    funct();

    for (int i = 0; i < 10; ++i) {
        cout << "Main thread is running!" << endl;
        sleep(1);
    }

    cout << "Bye!" << endl;
    return 0;
}

void funct() {
    int id = 1;
    int count = 10;
    string str = "Hello World";
    Show s1(id, count, str);
    thread t1(s1);
    
    thread_guard gt(t1);

    int n1, n2;
    cout << "Please enter two numbers: " << endl;
    cin >> n1 >> n2;

    try {
        if (n2 == 0) {
            // 子线程出现异常
            throw runtime_error("n2 can't be 0");
        }
        cout << "n1 / n2 = " << n1 / n2 << endl;
    } catch(runtime_error) {
        cout << "Quit with exception" << endl;
        // 结束后会自动执行析构函数
        // t1.join();
        // 主线程结束
        // return;
    }
    // 结束后会自动执行析构函数
    // t1.join();
}

传递引用

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
#include <iostream>
#include <thread>

using namespace std;

// void threadFunction(int n);
// void threadFunction(int *pt);
void threadFunction(int &r);

int main() {
    int i = 0;
    // thread t1(threadFunction, i);
    // thread t1(threadFunction, &i);
    // 子线程修改主线程的变量
    // 按引用的方式传递参数, 此时 i 不是副本了，而是 i 的引用
    // 传递字符串的字面值的话是按照指针进行传递 char const *, 会发生隐式转换,可以不需要ref()
    thread t1(threadFunction, ref(i));
    t1.join();
    cout << "i = " << i << endl;

    return 0;
}

// void threadFunction(int n) {
// void threadFunction(int *pt) {
void threadFunction(int &r) {
    r += 100;
}

CPU核心数

cout << thread::hardware_concurrency() << endl 查看cpu核心数

ps -eLF 查看线程运行在哪个核心上

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
#include <iostream>
#include <string>
#include <thread>
#include <unistd.h>

using namespace std;

void show(int id, int count, const string &str);

int main() {
    // 查看cpu核心数
    cout << thread::hardware_concurrency() << endl;
    // 创建线程, 传递参数
    thread t1(show, 1, 100, "hello");
    thread t2(show, 2, 100, "world");
    thread t3(show, 3, 100, "good");
    thread t4(show, 4, 100, "hcjjj");
    // 主线程阻塞，等待子线程运行结束
    t1.join();
    t2.join();
    t3.join();
    t4.join();
    cout << "Bye!" << endl;
    return 0;
}

void show(int id, int count, const string &str) {
    for (int i = 0; i < count; ++i) {
        cout << "Thread " << id << ": " << str << endl;
        sleep(1);
    }
}

多线程应用

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
#include <iostream>
#include <numeric>
#include <thread>
#include <vector>

using namespace std;

int parallel_accumulate(vector<int>::iterator first, vector<int>::iterator last, int init);
void accumulate_block(vector<int>::iterator first, vector<int>::iterator last, int &sum);

int main() {
    vector<int> ivec(100);
    for (int i = 1; i <= 100; ++i) {
        ivec.push_back(i);
    }
    cout << parallel_accumulate(ivec.begin(), ivec.end(), 0) << endl;
    return 0;
}

int parallel_accumulate(vector<int>::iterator first, vector<int>::iterator last, int init) {
    // 元素数量
    const unsigned int length = distance(first, last);
    if (!length) return init;
    // 每个线程处理25个数
    const unsigned int min_per_thread = 25;
    // 需要多少个线程 理论和硬件取最小值
    const unsigned int max_threads = (length + min_per_thread - 1) / min_per_thread;
    const unsigned int hardware_threads = thread::hardware_concurrency();
    const unsigned int num_threads = min((hardware_threads != 0 ? hardware_threads : 2), max_threads);
    const unsigned int block_size = length / num_threads;

    // 减去一个主线程, 主线程也参与进来
    vector<thread> threads(num_threads - 1);
    vector<int> results(num_threads);

    vector<int>::iterator block_start = first;
    // 分配子线程
    for (int i = 0; i < (num_threads - 1); ++i) {
        vector<int>::iterator block_end = block_start;
        // 迭代器向前调整 block_size 个位置
        advance(block_end, block_size);
        threads[i] = thread(accumulate_block, block_start, block_end, ref(results[i]));
        block_start = block_end;
    }
    for (auto &r : threads) {
        r.join();
    }
    // 主线程收尾
    accumulate_block(block_start, last, results[num_threads - 1]);
    return accumulate(results.begin(), results.end(), init);
}

void accumulate_block(vector<int>::iterator first, vector<int>::iterator last, int &sum) {
    sum = accumulate(first, last, sum);
}

使用互斥

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
#include <iostream>
#include <thread>
#include <mutex>
#include <vector>
#include <unistd.h>

using namespace std;

void write_to_vector(int id, int value);

vector<int> ivec;
mutex guard_mutex;

int main() {
    thread t1(write_to_vector, 1, 100);
    thread t2(write_to_vector, 2, 200);
    thread t3(write_to_vector, 3, 300);
    thread t4(write_to_vector, 4, 400);
    thread t5(write_to_vector, 5, 500);

    t1.join();
    t2.join();
    t3.join();
    t4.join();
    t5.join();

    for (int i = 0; i < ivec.size(); ++i) {
        cout << ivec[i] << " ";
    }
    return 0;
}

void write_to_vector(int id, int value) {
    cout << "Thread " << id << " is running" << " value: " << value << endl; 
    // 使用互斥锁初始化 lock_guard
    // RAII 机制
    lock_guard<mutex> guard(guard_mutex);
    cout << "Thread " << id << " lock to the resource" << endl;
    ivec.push_back(value);
    sleep(1);
}

游离指针

不要将受保护的数据，以指针或者引用的方式传递到所保护范围之外

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
#include <iostream>
#include <thread>
#include <mutex>
#include <unistd.h>

using namespace std;

class User_data {
public:
    int a;
    char c;
    void change_data() {
        a *= 10;
        c += 10;
    }
};

// 游离指针
User_data *unprotect_pt;
void function(User_data &my_data) {
    unprotect_pt = &my_data;
}

class Protect_data {
private:
    User_data data;
    mutex guard_mutex;
public:
    Protect_data() {
        data.a = 1;
        data.c = 'A';
    }
    void process_data() {
        // 加锁
        lock_guard<mutex> guard(guard_mutex);
        data.a += 10;
        data.c += 1;
        // 受保护的data 被一个外部的一个游离指针捕获(禁止这么干⚠️)
        function(data);
        sleep(2);
    }
    void print() {
        cout << "data.a = " << data.a << ", data.c = " << data.c << endl;
    }

};

void thred_function(Protect_data &pd) {
    // 修改pd中的数据
    pd.process_data();
}

int main() {
    Protect_data pd;
    pd.print();
    // 有加锁 所以 t1 t2 不能同时进行修改
    thread t1(thred_function, ref(pd));
    thread t2(thred_function, ref(pd));
    t1.join();
    t2.join();
    pd.print();
    // 游离的指针改变了受保护的数据
    unprotect_pt->change_data();
    pd.print();
    return 0;
}

死锁的产生

情况1

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
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>
// linux 下的一个库
#include <unistd.h>

using namespace std;

void thread1();
void thread2();

mutex mt;
int num = 100;

int main() {
    thread t1(thread1);
    thread t2(thread2);
    t1.join();
    t2.join();
    cout << "All threads end." << endl;

    return 0;
}

void thread1() {
    cout << "Thread1 is runnning: " << endl;
    lock_guard<mutex> guard(mt);
    cout << "Thread1: Shared date ---> num = " << num << endl;
    // 一直占据资源
    while(1);
}

void thread2() {
    cout << "Thread2 is runnning: " << endl;
    // linux 的
    // sleep(1);
    // c++ 的 
    this_thread::sleep_for(chrono::seconds(1));
    lock_guard<mutex> guard(mt);
    cout << "Thread2: Shared date ---> num = " << num << endl;

}

情况2

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
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>
// linux 下的一个库
#include <unistd.h>

using namespace std;

void thread1();
void thread2();

mutex mt1;
mutex mt2;
int a = 100;
int b = 200;

int main() {
    thread t1(thread1);
    thread t2(thread2);
    // 主线程阻塞等待子线程运行结束
    t1.join();
    t2.join();
    // 子线程因为产生了死锁也互相阻塞
    cout << "All threads end." << endl;
    return 0;
}

void thread1() {
    cout << "Thread1 is runnning: " << endl;
    lock_guard<mutex> guard1(mt1);
    cout << "Thread1: Shared date ---> a = " << a << endl;
    sleep(1);
    lock_guard<mutex> guard2(mt2);
    cout << "Thread1: Shared date ---> b = " << b << endl;
    cout << "Thread1: Get shared data: a + b = " << a + b << endl;
}

void thread2() {
    cout << "Thread2 is runnning: " << endl;
    lock_guard<mutex> guard2(mt2);
    cout << "Thread2: Shared date ---> b = " << b << endl;
    sleep(1);
    lock_guard<mutex> guard1(mt1);
    cout << "Thread2: Shared date ---> a = " << b << endl;
    cout << "Thread2: Get shared data: a + b = " << a + b << endl;
}

避免的死锁

lock + lock_guard

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
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>
// linux 下的一个库
#include <unistd.h>

using namespace std;

void thread1();
void thread2();

mutex mt1;
mutex mt2;
int a = 100;
int b = 200;

int main() {
    thread t1(thread1);
    thread t2(thread2);
    // 主线程阻塞等待子线程运行结束
    t1.join();
    t2.join();
    // 子线程因为产生了死锁也互相阻塞
    cout << "All threads end." << endl;
    return 0;
}

void thread1() {
    cout << "Thread1 is runnning: " << endl;
    // 同时加锁
    lock(mt1, mt2);
    // 使用lock_guard在析构函数中自动解锁, adopt_lock检查是否上锁
    lock_guard<mutex> guard1(mt1, adopt_lock);
    lock_guard<mutex> guard2(mt2, adopt_lock);
    cout << "Thread1: Shared date ---> a = " << a << endl;
    sleep(1);
    cout << "Thread1: Shared date ---> b = " << b << endl;
    cout << "Thread1: Get shared data: a + b = " << a + b << endl;
}

void thread2() {
    cout << "Thread2 is runnning: " << endl;
    lock(mt1, mt2);
    lock_guard<mutex> guard1(mt1, adopt_lock);
    lock_guard<mutex> guard2(mt2, adopt_lock);
    cout << "Thread2: Shared date ---> b = " << b << endl;
    sleep(1);
    cout << "Thread2: Shared date ---> a = " << b << endl;
    cout << "Thread2: Get shared data: b - a = " << b - a << endl;
}

unique_lock + lock

unique_lock的使用

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
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>

using namespace std;

void thread1(int id, const string &str);

mutex mt;

int main() {
    thread t1(thread1, 1, "Hello world");
    // this_thread::sleep_for(chrono::seconds(1));
    t1.join();
    return 0;
}

// void thread1(int id, const string &str) {
//     lock_guard<mutex> guard(mt);
//     cout << "Thread" << id << ": " << str << endl;
// }

void thread1(int id, const string &str) {
    // 在构造函数中不自动加锁
    unique_lock<mutex> guard(mt, defer_lock);
    // ...
    // 手动加锁
    guard.lock();
    cout << "Thread" << id << ": " << str << endl;
    // 手动解锁
    guard.unlock();
    int sum = 0;
    for (int i = 0; i < 100; ++i) {
        sum += i;
    }
    guard.lock();
    cout << "Thread" << id << ": " << sum << 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>

using namespace std;

void thread1();
void thread2();

mutex mt1;
mutex mt2;
int a = 100;
int b = 200;

int main() {
    thread t1(thread1);
    thread t2(thread2);
    // 主线程阻塞等待子线程运行结束
    t1.join();
    t2.join();
    // 子线程因为产生了死锁也互相阻塞
    cout << "All threads end." << endl;
    return 0;
}

void thread1() {
    unique_lock<mutex> guard1(mt1, defer_lock);
    unique_lock<mutex> guard2(mt2, defer_lock);
    // guard1.lock();
    // guard2.lock();
    lock(guard1, guard2);
    cout << "Thread1 is runnning: " << endl;
    cout << "Thread1: Shared date ---> a = " << a << endl;
    this_thread::sleep_for(chrono::seconds(1));
    cout << "Thread1: Shared date ---> b = " << b << endl;
    cout << "Thread1: Get shared data: a + b = " << a + b << endl;
}

void thread2() {
    unique_lock<mutex> guard1(mt1, defer_lock);
    unique_lock<mutex> guard2(mt2, defer_lock);
    lock(guard1, guard2);
    cout << "Thread2 is runnning: " << endl;
    cout << "Thread2: Shared date ---> b = " << b << endl;
    this_thread::sleep_for(chrono::seconds(1));
    cout << "Thread2: Shared date ---> a = " << b << endl;
    cout << "Thread2: Get shared data: b - a = " << b - a << endl;
}

scoped_lock

c++17

g++ lesson20.cpp -pthread -std=c++17 && ./a.out

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
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>

using namespace std;

void thread1();
void thread2();

mutex mt1;
mutex mt2;
int a = 100;
int b = 200;

int main() {
    thread t1(thread1);
    thread t2(thread2);
    // 主线程阻塞等待子线程运行结束
    t1.join();
    t2.join();
    // 子线程因为产生了死锁也互相阻塞
    cout << "All threads end." << endl;
    return 0;
}

void thread1() {
    cout << "Thread1 is runnning: " << endl;
    // scoped_lock<mutex, mutex> guard(mt1, mt2);
    // C++ 17类模板参数自动推导
    scoped_lock guard(mt1, mt2);
    cout << "Thread1: Shared date ---> a = " << a << endl;
    this_thread::sleep_for(chrono::seconds(1));
    cout << "Thread1: Shared date ---> b = " << b << endl;
    cout << "Thread1: Get shared data: a + b = " << a + b << endl;
}

void thread2() {
    cout << "Thread2 is runnning: " << endl;
    // scoped_lock<mutex, mutex> guard(mt1, mt2);
    scoped_lock guard(mt1, mt2);
    cout << "Thread2: Shared date ---> b = " << b << endl;
    this_thread::sleep_for(chrono::seconds(1));
    cout << "Thread2: Shared date ---> a = " << b << endl;
    cout << "Thread2: Get shared data: a + b = " << a + b << 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
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
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>

using namespace std;

void thread1();
void thread2();
void thread3();

int count = -1;
mutex mtx;
condition_variable flag;

int main() {
    thread t1(thread1);
    thread t2(thread2);
    thread t3(thread3);
    t1.join();
    t2.join();
    t3.join();

    return 0;
}

void thread1() {
    int i = 0;
    this_thread::sleep_for(chrono::seconds(1));
    while (1) {
        // {
        //     lock_guard<mutex> lck(mtx);
        //     // count = i++;
        //     count = i;
        //     ++i;
        //     // RAII自动释放锁
        // }
        unique_lock<mutex> lck(mtx);
        count = i++;
        lck.unlock();
        // 发生信号 给一个线程
        // flag.notify_one();
        flag.notify_all();
        this_thread::sleep_for(chrono::seconds(1));
    }
}

void thread2() {
    while(1) {
        // lock_guard无法暂时释放锁
        // lock_guard<mutex> lck(mtx);
        unique_lock<mutex> lck(mtx);
        cout << "Thread2 waits for count: ";
        // 等待信号 会自动暂时释放锁
        flag.wait(lck);
        cout << count << endl;
    }
}

void thread3() {
    while(1) {
        // lock_guard无法暂时释放锁
        // lock_guard<mutex> lck(mtx);
        unique_lock<mutex> lck(mtx);
        cout << "Thread3 waits for count: ";
        // 等待信号 会自动暂时释放锁
        flag.wait(lck);
        cout << count << 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
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
 #include <chrono>
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>

using namespace std;
void consumer();
void producer(int n);

mutex mtx;
condition_variable produce, consume;

int cargo = -1;

int main() {
    thread consumers[10], producers[10];
    for (int i = 0; i < 10; ++i) {
        consumers[i] = thread(consumer);
        producers[i] = thread(producer, i + 1);
    }
    for (int i = 0; i < 10; ++i) {
        consumers[i].join();
        producers[i].join();
    }
    return 0;
}

void consumer() {
    unique_lock<mutex> lck(mtx);
    if (cargo == -1) {
        cout << "Consumer waits for cargo!" << endl;
        // 等待生产者生产
        consume.wait(lck);
    }
    cout << cargo << endl;
    cargo = -1;
    // 通知生产者进行生产
    produce.notify_one();

}
void producer(int n) {
    this_thread::sleep_for(chrono::seconds(5));
    unique_lock<mutex> lck(mtx);
    if (cargo != -1) {
        produce.wait(lck);
    }
    cargo = n;
    consume.notify_one();
}

condition_variable

wait() 函数的第二种使用方式

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
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <thread>

using namespace std;

mutex mtx;
condition_variable flag;
int count = -1;

void signal();
void wait(int n);

int main() {
    thread t1(signal);
    thread t2(wait, 2);
    thread t3(wait, 3);
    t1.join();
    t2.join();
    t3.join();

    return 0;
}

void signal() {
    this_thread::sleep_for(chrono::seconds(1));
    unique_lock<mutex> lck(mtx);
    cout << "notify all threads" << endl;
    lck.unlock();
    flag.notify_all();

    this_thread::sleep_for(chrono::seconds(1));
    lck.lock();
    count = 1;
    cout << "notify all threads again" << endl;
    lck.unlock();
    flag.notify_all();


}

void wait(int id) {
    unique_lock<mutex> lck(mtx);
    cout << "Thread" << id << " waits for count: " << endl;
    // wait 的第二个参数[需要满足的条件]
    flag.wait(lck, [](){return count == 1;});
    cout << "Thread " << id << ": " << count << endl;
}

创建异步任务

async

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
#include <iostream>
#include <future>
#include <chrono>
#include <thread>

using namespace std;
bool is_prime(int x);

int main() {
    // 方式一
    future<bool> fut = async(is_prime, 313222313);
    cout << "Checking whethe 313222313 is prime." << endl;
    // 等待结果
    bool ret = fut.get();
    if (ret) cout << "It is prime" << endl;
    else cout << "It is not prime" << endl;
    return 0;
}

bool is_prime(int x) {
    cout << "Calculating. Please wait ... " << endl;
    this_thread::sleep_for(chrono::seconds(5));
    for (int i = 2; i < x; ++i) {
        if (x % i == 0) return false;
    }
    return true;
}
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
#include <iostream>
#include <future>
#include <chrono>
#include <thread>

using namespace std;

class Prime {
public:
    bool is_prime(int x) {
        cout << "Calculating. Please wait ... " << endl;
        this_thread::sleep_for(chrono::seconds(5));
        for (int i = 2; i < x; ++i) {
            if (x % i == 0) return false;
        }
        return true;
    }
};

int main() {
    // 方式二
    Prime p;
    future<bool> fut = async(&Prime::is_prime, p, 31322313);
    cout << "Checking whethe 313222313 is prime." << endl;
    bool ret = fut.get();
    if (ret) cout << "It is prime" << endl;
    else cout << "It is not prime" << endl;
    return 0;
}

async选项

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
#include <iostream>
#include <future>
#include <chrono>
#include <thread>

using namespace std;
bool is_prime(int x);

int main() {
    // lacunch::async 异步线程
    // launch:deferred 同步线程(线程id相同， 只有一个线程)
    // future<bool> fut = async(std::launch::async, is_prime, 313222313);
    future<bool> fut = async(std::launch::deferred, is_prime, 313222313);
    // future<bool> fut = async(launch::deferred|launch::async, is_prime, 313222313);
    cout << "Checking whethe 313222313 is prime." << endl;
    cout << "main() thread id = " << std::this_thread::get_id() << endl;
    this_thread::sleep_for(chrono::seconds(5));
    // 子线程传递数据给主线程
    bool ret = fut.get();
    if (ret) cout << "It is prime" << endl;
    else cout << "It is not prime" << endl;
    return 0;
}

bool is_prime(int x) {
    cout << "Calculating. Please wait ... " << endl;
    cout << "is_prime() thread id = " << std::this_thread::get_id() << endl;
    this_thread::sleep_for(chrono::seconds(5));
    for (int i = 2; i < x; ++i) {
        if (x % i == 0) return false;
    }
    return true;
}

promise

通过promise对象实现主线程与子线程、子线程与子线程之间数据的传输

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
#include <chrono>
#include <iostream>
#include <future>
#include <thread>

using namespace std;

void add_function(promise<int> &, int , int);

int main() {
    promise<int> pm;
    future<int> future;
    // make future promise get relationship
    future = pm.get_future();
    thread t1(add_function, ref(pm), 10, 20);
    // wait for child thread finish, blocked function
    int sum = future.get();
    cout << "sum = " << sum << endl;
    t1.join();
    return 0;
}

void add_function(promise<int> &mypromise, int x, int y) {
    cout << "x = " << x << ", y = " << y << endl;
    int sum = 0;
    sum = x + y;
    this_thread::sleep_for(chrono::seconds(3));
    mypromise.set_value(sum);
}

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
#include <iostream>
#include <future>
#include <thread>

using namespace std;

void thread1(promise<int> &mypromise, int n);
void thread2(future<int> &fut);

int main() {
    promise<int> pm;
    future<int> future;
    future = pm.get_future();
    // 主线程通过 pm 传递数据给 t1
    thread t1(thread1, ref(pm), 5);
    // t1 通过 future 传递给 t2
    thread t2(thread2, ref(future));
    t1.join();
    t2.join();

    return 0;
}

void thread1(promise<int> &mypromise, int n) {
    cout << "thread1 input value: " << n << endl;
    n *= 100;
    this_thread::sleep_for(chrono::seconds(3));
    mypromise.set_value(n);
}

void thread2(future<int> &fut) {
     cout << "thread2 get value: " << fut.get() << endl;
}

packaged_task

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
#include <iostream>
#include <future>
#include <thread>

using namespace std;

int add_func(int, int);

int main() {
    cout << "main() thread id = " << this_thread::get_id() << endl;
    // 不需要对可调用对象进行过多的修改
    // 可调用对象 -> 异步线程
    // packaged_task<int(int, int)> ptk([](int x, int y){
    //     cout << "child() thread id = " << this_thread::get_id() << endl;
    //     cout << "x = " << x << ", y = " << y << endl;
    //     return x + y;
    // });
    packaged_task<int(int, int)> ptk(add_func);
    future<int> future;
    future = ptk.get_future();
    thread t(std::move(ptk), 10, 20);
    int result = future.get();
    cout << "Add result = " << result << endl;
    t.join();

    return 0;
}

int add_func(int x, int y) {
    cout << "child() thread id = " << this_thread::get_id() << endl;
    cout << "x = " << x << ", y = " << y << endl;
    return x + y;
}

限时等待

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
#include <chrono>
#include <iostream>
#include <future>
#include <thread>

using namespace std;

void thread1(promise<int> &mypromise, int n);
void thread2(future<int> &fut);

int main() {
    promise<int> pm;
    future<int> future;
    future = pm.get_future();
    // 主线程通过 pm 传递数据给 t1
    thread t1(thread1, ref(pm), 5);
    // t1 通过 future 传递给 t2
    thread t2(thread2, ref(future));
    t1.join();
    t2.join();

    return 0;
}

void thread1(promise<int> &mypromise, int n) {
    cout << "thread1 input value: " << n << endl;
    n *= 100;
    // 需要20s
    this_thread::sleep_for(chrono::seconds(20));
    mypromise.set_value(n);
}

void thread2(future<int> &fut) {
    chrono::milliseconds span(10000);
    // 限时等待 10s
    if (fut.wait_for(span) == future_status::timeout) {
        cout << "Time out" << endl;
        cout << "Thread2 cannot wait any more!" << endl;
        return;
    }
    cout << "thread2 get value: " << fut.get() << 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
#include <iostream>
#include <ratio>
#include <thread>
#include <chrono>

using namespace std;

void do_something() {
    // 计时数据为 int 类型
    // 1/1s 为一个计时单位
    // 共5个计时单位 , 延时 5s
    this_thread::sleep_for(chrono::duration<int, ratio<1, 1>>(2));
    cout << "do something" << endl;
}

int main() {
    // 高精度时钟类
    auto start = chrono::high_resolution_clock::now();
    do_something();
    auto stop = chrono::high_resolution_clock::now();
    auto durations = chrono::duration<double, ratio<1, 1>>(stop - start).count();
    cout << "do_something takes " << durations << 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
29
30
31
32
33
34
35
36
37
38
39
40
#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>

using namespace std;

timed_mutex tmx;

void func1(int id, const string &str) {
    if (tmx.try_lock()) {
        cout << "Thread1 " << id << "locked and do something" << endl;
        for (int i = 0; i < 10; ++i) {
            cout << str << endl;
            this_thread::sleep_for(chrono::seconds(1));
        }
        tmx.unlock();
    }
}

void func2(int id, const string &str) {
    this_thread::sleep_for(chrono::seconds(1));
    // 尝试等待5s
    if (tmx.try_lock_for(chrono::seconds(5))) {
        cout << "Thread " << id << ": " << str << endl;
        tmx.unlock();
    } else {
        cout << "Thread2 cannot wait anymore";
    }
    
}

int main() {
    thread t1(func1, 1, "Hello world");
    thread t2(func2, 2, "Gooooood");
    t1.join();
    t2.join();

    return 0;
}

atomic原子操作

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
#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>
#include <atomic>

using namespace std;

void my_thread();
int count = 0;
mutex my_mutex;
atomic<int> count2(0);

int main() {
    thread t1(my_thread);
    thread t2(my_thread);
    auto start = chrono::high_resolution_clock::now();
    t1.join();
    t2.join();
    auto stop = chrono::high_resolution_clock::now();
    auto durations = chrono::duration<double, ratio<1, 1>>(stop - start).count();
    cout << "durations = " << durations << endl;
    // cout << "count = " << count << endl;
    cout << "count = " << count2 << endl;
    
    return 0;
}

void my_thread() {
    for (int j = 0; j < 5; ++j) {
        for (int i = 0; i < 10000000; ++i) {
            // 加锁解锁 消耗资源
            // my_mutex.lock();
            // ++count;
            // my_mutex.unlock();
            ++count2;
        }
    }
}
快来参与讨论吧