std::thread
void print(int n, const std::string &str) {
std::string msg = std::to_string(n) + " : " + str;
std::cout << msg << std::endl;
}
int main() {
std::vector<std::string> s = {
"Educative.blog",
"Educative",
"courses",
"are great"
};
std::vector<std::thread> threads;
for (int i = 0; i < s.size(); i++) {
threads.push_back(std::thread(print, i, s[i]));
}
for (auto &th : threads) {
th.join();
}
return 0;
}
ex2:
#include <iostream>
#include <utility>
#include <thread>
#include <functional>
void f1(int n)
{
for (int i = 0; i < 5; ++i) {
std::cout << "Thread " << n << " executing\n";
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
void f2(int& n)
{
for (int i = 0; i < 5; ++i) {
std::cout << "Thread 2 executing\n";
++n;
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
int main()
{
int n = 0;
std::thread t1; // t1 is not a thread
std::thread t2(f1, n + 1); // pass by value
std::thread t3(f2, std::ref(n)); // pass by reference
std::thread t4(std::move(t3)); // t4 is now running f2(). t3 is no longer a thread
t2.join();
t4.join();
std::cout << "Final value of n is " << n << '\n';
}
std::atomic
// global shared data
std::atomic<bool> flag = false; // #include <atomic>
thread1() {
flag = false;
Type* value = new Type(/* parameters */);
thread2(value);
while (true) {
if (flag == true) {
apply(value);
break;
}
}
thread2.join();
if (nullptr != value) { delete value; }
return;
}
thread2(Type* value) {
// do some evaluations
value->update(/* parameters */);
flag = true;
return;
}
std::mutex
std::mutex 是C++11 中最基本的互斥量，std::mutex 对象提供了独占所有权的特性——即不支持递归地对 std::mutex 对象上锁，而 std::recursive_lock 则可以递归地对互斥量对象上锁。
Mutex 系列类(四种)
- std::mutex，最基本的 Mutex 类。
- std::recursive_mutex，递归 Mutex 类。
- std::time_mutex，定时 Mutex 类。
- std::recursive_timed_mutex，定时递归 Mutex 类。
Lock 类（两种）
- std::lock_guard，与 Mutex RAII 相关，方便线程对互斥量上锁。
- std::unique_lock，与 Mutex RAII 相关，方便线程对互斥量上锁，但提供了更好的上锁和解锁控制。
Resource Acquisition Is Initialization : RAII
ex :
volatile int counter(0); // non-atomic counter
std::mutex mtx; // locks access to counter
void attempt_10k_increases() {
for (int i=0; i<10000; ++i) {
if (mtx.try_lock()) { // only increase if currently not locked:
++counter;
mtx.unlock();
}
}
}
int main (int argc, const char* argv[]) {
std::thread threads[10];
for (int i=0; i<10; ++i)
threads[i] = std::thread(attempt_10k_increases);
for (auto& th : threads) th.join();
std::cout << counter << " successful increases of the counter.\n";
return 0;
}
std::lock_guard
lock_guard 对象并不负责管理 Mutex 对象的生命周期，lock_guard 对象只是简化了 Mutex 对象的上锁和解锁操作，方便线程对互斥量上锁，即在某个 lock_guard 对象的声明周期内，它所管理的锁对象会一直保持上锁状态；而 lock_guard 的生命周期结束之后，它所管理的锁对象会被解锁。
ex:
std::mutex mtx;
void print_even (int x) {
if (x%2==0) std::cout << x << " is even\n";
else throw (std::logic_error("not even"));
}
void print_thread_id (int id) {
try {
// using a local lock_guard to lock mtx guarantees unlocking on destruction / exception:
std::lock_guard<std::mutex> lck (mtx);
print_even(id);
}
catch (std::logic_error&) {
std::cout << "[exception caught]\n";
}
}
int main ()
{
std::thread threads[10];
// spawn 10 threads:
for (int i=0; i<10; ++i)
threads[i] = std::thread(print_thread_id,i+1);
for (auto& th : threads) th.join();
return 0;
}
std::unique_lock
unique_lock 对象以独占所有权的方式（ unique owership）管理 mutex 对象的上锁和解锁操作，所谓独占所有权，就是没有其他的 unique_lock 对象同时拥有某个 mutex 对象的所有权。
unique_lock 对象同样也不负责管理 Mutex 对象的生命周期，unique_lock 对象只是简化了 Mutex 对象的上锁和解锁操作，方便线程对互斥量上锁，即在某个 unique_lock 对象的声明周期内，它所管理的锁对象会一直保持上锁状态；而 unique_lock 的生命周期结束之后，它所管理的锁对象会被解锁，这一点和 lock_guard 类似.
std::mutex foo, bar;
void task_a() { std::lock(foo, bar); // simultaneous lock (prevents deadlock)
std::unique_lock < std::mutex > lck1(foo, std::adopt_lock);
//adopt 收养一个lock ： 构造unique_lock时候就要lock mtx
std::unique_lock < std::mutex > lck2(bar, std::adopt_lock);
std::cout << "task a\n";
// (unlocked automatically on destruction of lck1 and lck2)
}
void task_b() { // foo.lock(); bar.lock(); // replaced by:
std::unique_lock < std::mutex > lck1, lck2;
lck1 = std::unique_lock < std::mutex > (bar, std::defer_lock);
//构建unique_lock的时候不去lock
lck2 = std::unique_lock < std::mutex > (foo, std::defer_lock);
std::lock(lck1, lck2);
// simultaneous lock (prevents deadlock)
std::cout << "task b\n";
// (unlocked automatically on destruction of lck1 and lck2)
}
int main() { std::thread th1(task_a);
std::thread th2(task_b);
th1.join();
th2.join();
return 0;
}
You can lock and unlock a std::unique_lock. std::lock_guard will be locked only once on construction and unlocked on destruction.
By default std::lock_guard and std::unique_lock are the same.
However, std::unique_lock might have a tad more overhead.
Note that these days (since, C++17) one should use std::scoped_lock instead of std::lock_guard.
std::future
std::promise
promise 对象可以保存某一类型 T 的值，该值可被 future 对象读取（可能在另外一个线程中），因此 promise 也提供了一种线程同步的手段。
void print_int(std::future<int>& fut) {
int x = fut.get(); // 获取共享状态的值.
std::cout << "value: " << x << '\n'; // 打印 value: 10.
}
int main ()
{
std::promise<int> prom; // 生成一个 std::promise<int> 对象.
std::future<int> fut = prom.get_future(); // 和 future 关联.
std::thread t(print_int, std::ref(fut)); // 将 future 交给另外一个线程t.
prom.set_value(10); // 设置共享状态的值, 此处和线程t保持同步.
t.join();
return 0;
}
ex2:
std::promise<int> prom;
void print_global_promise () {
std::future<int> fut = prom.get_future();
int x = fut.get();
std::cout << "value: " << x << '\n';
}
int main ()
{
std::thread th1(print_global_promise);
prom.set_value(10);
th1.join();
prom = std::promise<int>(); // prom 被move赋值为一个新的 promise 对象. 不做这一步的话，下边的promise access 会报错 ： promise 已经satisfied了
std::thread th2 (print_global_promise);
prom.set_value (20);
th2.join();
return 0;
}
thread 要执行的任务，如何返回一个值？那这时候就要用 packages_task来把function 包装成task
std::packaged_task
可以通过 std::packged_task::get_future 来获取与共享状态相关联的 std::future 对象。在调用该函数之后，两个对象共享相同的共享状态，具体解释如下：
- std::packaged_task 对象是异步 Provider，它在某一时刻通过调用被包装的任务来设置共享状态的值。
- std::future 对象是一个异步返回对象，通过它可以获得共享状态的值，当然在必要的时候需要等待共享状态标志变为 ready.
ex：
// count down taking a second for each value:
int countdown (int from, int to) {
for (int i=from; i!=to; --i) {
std::cout << i << '\n';
std::this_thread::sleep_for(std::chrono::seconds(1));
}
std::cout << "Finished!\n";
return from - to;
}
int main ()
{
std::packaged_task<int(int,int)> task(countdown); // 设置 packaged_task
std::future<int> ret = task.get_future(); // 获得与 packaged_task 共享状态相关联的 future 对象.
std::thread th(std::move(task), 10, 0); //创建一个新线程完成计数任务.
int value = ret.get(); // 等待任务完成并获取结果.
std::cout << "The countdown lasted for " << value << " seconds.\n";
th.join();
return 0;
}
std::async
一个有效(valid)的 std::future 对象通常由以下三种 Provider 创建
std::async 函数
std::promise::get_future，
std::packaged_task::get_future
一个 std::future 对象只有在有效(valid)的情况下才有用(useful)，由 std::future 默认构造函数创建的 future 对象不是有效的（除非当前非有效的 future 对象被 move 赋值另一个有效的 future 对象）。
int do_get_value() { return 10; }
int main ()
{
std::future<int> fut = std::async(do_get_value);
std::shared_future<int> shared_fut = fut.share();
// 共享的 future 对象可以被多次访问.
std::cout << "value: " << shared_fut.get() << '\n';
std::cout << "its double: " << shared_fut.get()*2 << '\n';
return 0;
}
ex：
int main ()
{
// 由默认构造函数创建的 std::future 对象,
// 初始化时该 std::future 对象处于为 invalid 状态.
std::future<int> foo, bar;
foo = std::async(do_get_value); // move 赋值, foo 变为 valid.
bar = std::move(foo); // move 赋值, bar 变为 valid, 而 move 赋值以后 foo 变为 invalid.
if (foo.valid())
std::cout << "foo's value: " << foo.get() << '\n';
else
std::cout << "foo is not valid\n";
if (bar.valid())
std::cout << "bar's value: " << bar.get() << '\n';
else
std::cout << "bar is not valid\n";
return 0;
}
| launch::async | Asynchronous:
异步任务会在另外一个线程中调用，
并通过共享状态返回异步任务的结果
（一般是调用 std::future::get() 获取异步任务的结果）。 |
| launch::deferred | Deferred:
异步任务将会在共享状态被访问时调用，
相当与按需调用（即延迟(deferred)调用）。 |
void
do_print_ten(char c, int ms)
{
for (int i = 0; i < 10; ++i) {
std::this_thread::sleep_for(std::chrono::milliseconds(ms));
std::cout << c;
}
}
int
main()
{
std::cout << "with launch::async:\n";
std::future < void >foo =
std::async(std::launch::async, do_print_ten, '*', 100);
std::future < void >bar =
std::async(std::launch::async, do_print_ten, '@', 200);
// async "get" (wait for foo and bar to be ready):
foo.get();
bar.get();
std::cout << "\n\n";
std::cout << "with launch::deferred:\n";
foo = std::async(std::launch::deferred, do_print_ten, '*', 100);
bar = std::async(std::launch::deferred, do_print_ten, '@', 200);
// deferred "get" (perform the actual calls):
foo.get();//调用get时才会执行function
bar.get();
std::cout << '\n';
return 0;
}
std::future::wait()
等待与当前std::future 对象相关联的共享状态的标志变为 ready.
std::condition_variable
当 std::condition_variable 对象的某个 wait 函数被调用的时候，它使用 std::unique_lock(通过 std::mutex) 来锁住当前线程。当前线程会一直被阻塞，直到另外一个线程在相同的 std::condition_variable 对象上调用了 notification 函数来唤醒当前线程。
example :
std::mutex mtx; // 全局互斥锁.
std::condition_variable cv; // 全局条件变量.
bool ready = false; // 全局标志位.
void do_print_id(int id)
{
std::unique_lock <std::mutex> lck(mtx);
while (!ready) // 如果标志位不为 true, 则等待...
cv.wait(lck); // 当前线程被阻塞, 当全局标志位变为 true 之后,
// 线程被唤醒, 继续往下执行打印线程编号id.
std::cout << "thread " << id << '\n';
}
void go()
{
std::unique_lock <std::mutex> lck(mtx);
ready = true; // 设置全局标志位为 true.
cv.notify_all(); // 唤醒所有线程.
}
int main()
{
std::thread threads[10];
// spawn 10 threads:
for (int i = 0; i < 10; ++i)
threads[i] = std::thread(do_print_id, i);
std::cout << "10 threads ready to race...\n";
go(); // go!
for (auto & th:threads)
th.join();
return 0;
}
ex:
// global shared data
std::mutex m; // #include <mutex>
std::condition_variable cv; // #include <condition_variable>
bool flag = false;
thread1() {
flag = false;
Type* value = new Type(/* parameters */);
thread2(value);
std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [](){ return flag; });
apply(value);
lk.unlock();
thread2.join();
if (nullptr != value) { delete value; }
return;
}
thread2(Type* value) {
std::lock_guard<std::mutex> lk(m);
// do some evaluations
value->update(/* parameters */);
flag = true;
cv.notify_one();
return;
}
Ref：
https://www.educative.io/blog/modern-multithreading-and-concurrency-in-cpp
https://www.cnblogs.com/haippy/p/3284540.html
