【C/C++开发】C++11 并发指南三(std::mutex 详解)

<mutex> 头文件介绍

Mutex 系列类(四种)

Lock 类（两种）

其他类型

函数

std::mutex 介绍

std::mutex 的成员函数

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex

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::recursive_mutex 介绍

std::time_mutex 介绍

#include <iostream>       // std::cout
#include <chrono>         // std::chrono::milliseconds
#include <thread>         // std::thread
#include <mutex>          // std::timed_mutex

std::timed_mutex mtx;

void fireworks() {
// waiting to get a lock: each thread prints "-" every 200ms:
while (!mtx.try_lock_for(std::chrono::milliseconds(200))) {
std::cout << "-";
}
// got a lock! - wait for 1s, then this thread prints "*"
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
std::cout << "*\n";
mtx.unlock();
}

int main ()
{
std::thread threads[10];
// spawn 10 threads:
for (int i=0; i<10; ++i)
threads[i] = std::thread(fireworks);

for (auto& th : threads) th.join();

return 0;
}

std::recursive_timed_mutex 介绍

std::lock_guard 介绍

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex, std::lock_guard
#include <stdexcept>      // std::logic_error

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 介绍

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex, std::unique_lock

std::mutex mtx;           // mutex for critical section

void print_block (int n, char c) {
// critical section (exclusive access to std::cout signaled by lifetime of lck):
std::unique_lock<std::mutex> lck (mtx);
for (int i=0; i<n; ++i) {
std::cout << c;
}
std::cout << '\n';
}

int main ()
{
std::thread th1 (print_block,50,'*');
std::thread th2 (print_block,50,'$');

th1.join();
th2.join();

return 0;
}

std::promise 类介绍

#include <iostream>       // std::cout
#include <functional>     // std::ref
#include <thread>         // std::thread
#include <future>         // std::promise, std::future

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;
}

std::promise 构造函数

promise();

template <class Alloc> promise (allocator_arg_t aa, const Alloc& alloc);

promise (const promise&) = delete;

promise (promise&& x) noexcept;

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <future>         // std::promise, std::future

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 对象.

std::thread th2 (print_global_promise);
prom.set_value (20);
th2.join();

return 0;
}

std::promise::get_future 介绍

std::promise::set_value 介绍

void set_value (const T& val);
void set_value (T&& val);

void promise<R&>::set_value (R& val);   // when T is a reference type (R&)
void promise<void>::set_value (void);   // when T is void

std::promise::set_exception 介绍

#include <iostream>       // std::cin, std::cout, std::ios
#include <functional>     // std::ref
#include <thread>         // std::thread
#include <future>         // std::promise, std::future
#include <exception>      // std::exception, std::current_exception

void get_int(std::promise<int>& prom) {
int x;
std::cout << "Please, enter an integer value: ";
std::cin.exceptions (std::ios::failbit);   // throw on failbit
try {
std::cin >> x;                         // sets failbit if input is not int
prom.set_value(x);
} catch (std::exception&) {
prom.set_exception(std::current_exception());
}
}

void print_int(std::future<int>& fut) {
try {
int x = fut.get();
std::cout << "value: " << x << '\n';
} catch (std::exception& e) {
std::cout << "[exception caught: " << e.what() << "]\n";
}
}

int main ()
{
std::promise<int> prom;
std::future<int> fut = prom.get_future();

std::thread th1(get_int, std::ref(prom));
std::thread th2(print_int, std::ref(fut));

th1.join();
th2.join();
return 0;
}

std::promise::set_value_at_thread_exit 介绍

std::promise::swap 介绍

#include <iostream>     // std::cout
#include <future>       // std::packaged_task, std::future
#include <chrono>       // std::chrono::seconds
#include <thread>       // std::thread, std::this_thread::sleep_for

// 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;
}

concurrency ) ./Packaged_Task1
10
9
8
7
6
5
4
3
2
1
Finished!
The countdown lasted for 10 seconds.

packaged_task() noexcept;

template <class Fn>
explicit packaged_task (Fn&& fn);

template <class Fn, class Alloc>
explicit packaged_task (allocator_arg_t aa, const Alloc& alloc, Fn&& fn);

packaged_task (const packaged_task&) = delete;

packaged_task (packaged_task&& x) noexcept;

#include <iostream>     // std::cout
#include <utility>      // std::move
#include <future>       // std::packaged_task, std::future
#include <thread>       // std::thread

int main ()
{
std::packaged_task<int(int)> foo; // 默认构造函数.

// 使用 lambda 表达式初始化一个 packaged_task 对象.
std::packaged_task<int(int)> bar([](int x){return x*2;});

foo = std::move(bar); // move-赋值操作，也是 C++11 中的新特性.

// 获取与 packaged_task 共享状态相关联的 future 对象.
std::future<int> ret = foo.get_future();

std::thread(std::move(foo), 10).detach(); // 产生线程，调用被包装的任务.

int value = ret.get(); // 等待任务完成并获取结果.
std::cout << "The double of 10 is " << value << ".\n";

return 0;
}

#include <iostream>     // std::cout
#include <utility>      // std::move
#include <future>       // std::packaged_task, std::future
#include <thread>       // std::thread

// 在新线程中启动一个 int(int) packaged_task.
std::future<int> launcher(std::packaged_task<int(int)>& tsk, int arg)
{
if (tsk.valid()) {
std::future<int> ret = tsk.get_future();
std::thread (std::move(tsk),arg).detach();
return ret;
}
else return std::future<int>();
}

int main ()
{
std::packaged_task<int(int)> tsk([](int x){return x*2;});

std::future<int> fut = launcher(tsk,25);

std::cout << "The double of 25 is " << fut.get() << ".\n";

return 0;
}

#include <iostream>     // std::cout
#include <utility>      // std::move
#include <future>       // std::packaged_task, std::future
#include <thread>       // std::thread

int main ()
{
std::packaged_task<int(int)> tsk([](int x) { return x * 3; })); // package task

std::future<int> fut = tsk.get_future();   // 获取 future 对象.

std::thread(std::move(tsk), 100).detach();   // 生成新线程并调用packaged_task.

int value = fut.get();                     // 等待任务完成, 并获取结果.

std::cout << "The triple of 100 is " << value << ".\n";

return 0;
}

#include <iostream>     // std::cout
#include <utility>      // std::move
#include <future>       // std::packaged_task, std::future
#include <thread>       // std::thread

// a simple task:
int triple (int x) { return x*3; }

int main ()
{
std::packaged_task<int(int)> tsk (triple); // package task

std::future<int> fut = tsk.get_future();
std::thread (std::move(tsk), 100).detach();
std::cout << "The triple of 100 is " << fut.get() << ".\n";

// re-use same task object:
tsk.reset();
fut = tsk.get_future();
std::thread(std::move(tsk), 200).detach();
std::cout << "Thre triple of 200 is " << fut.get() << ".\n";

return 0;
}

std::future 介绍

// future example
#include <iostream>             // std::cout
#include <future>               // std::async, std::future
#include <chrono>               // std::chrono::milliseconds

// a non-optimized way of checking for prime numbers:
bool
is_prime(int x)
{
for (int i = 2; i < x; ++i)
if (x % i == 0)
return false;
return true;
}

int
main()
{
// call function asynchronously:
std::future < bool > fut = std::async(is_prime, 444444443);

// do something while waiting for function to set future:
std::cout << "checking, please wait";
std::chrono::milliseconds span(100);
while (fut.wait_for(span) == std::future_status::timeout)
std::cout << '.';

bool x = fut.get();         // retrieve return value

std::cout << "\n444444443 " << (x ? "is" : "is not") << " prime.\n";

return 0;
}

std::future 成员函数

future() noexcept;

future (const future&) = delete;

future (future&& x) noexcept;

std::future<int> fut;           // 默认构造函数
fut = std::async(do_some_task);   // move-赋值操作。

#include <iostream>       // std::cout
#include <future>         // std::async, std::future, std::shared_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;
}

T get();

R& future<R&>::get();       // when T is a reference type (R&)

void future<void>::get();   // when T is void

#include <iostream>       // std::cin, std::cout, std::ios
#include <functional>     // std::ref
#include <thread>         // std::thread
#include <future>         // std::promise, std::future
#include <exception>      // std::exception, std::current_exception

void get_int(std::promise<int>& prom) {
int x;
std::cout << "Please, enter an integer value: ";
std::cin.exceptions (std::ios::failbit);   // throw on failbit
try {
std::cin >> x;                         // sets failbit if input is not int
prom.set_value(x);
} catch (std::exception&) {
prom.set_exception(std::current_exception());
}
}

void print_int(std::future<int>& fut) {
try {
int x = fut.get();
std::cout << "value: " << x << '\n';
} catch (std::exception& e) {
std::cout << "[exception caught: " << e.what() << "]\n";
}
}

int main ()
{
std::promise<int> prom;
std::future<int> fut = prom.get_future();

std::thread th1(get_int, std::ref(prom));
std::thread th2(print_int, std::ref(fut));

th1.join();
th2.join();
return 0;
}

#include <iostream>       // std::cout
#include <future>         // std::async, std::future
#include <utility>        // std::move

int do_get_value() { return 11; }

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;
}

#include <iostream>                // std::cout
#include <future>                // std::async, std::future
#include <chrono>                // std::chrono::milliseconds

// a non-optimized way of checking for prime numbers:
bool do_check_prime(int x) // 为了体现效果, 该函数故意没有优化.
{
for (int i = 2; i < x; ++i)
if (x % i == 0)
return false;
return true;
}

int main()
{
// call function asynchronously:
std::future < bool > fut = std::async(do_check_prime, 194232491);

std::cout << "Checking...\n";
fut.wait();

std::cout << "\n194232491 ";
if (fut.get()) // guaranteed to be ready (and not block) after wait returns
std::cout << "is prime.\n";
else
std::cout << "is not prime.\n";

return 0;
}

concurrency ) ./Future-wait
Checking...

194232491 is prime.
concurrency )

template <class Rep, class Period>
future_status wait_for (const chrono::duration<Rep,Period>& rel_time) const;

#include <iostream>                // std::cout
#include <future>                // std::async, std::future
#include <chrono>                // std::chrono::milliseconds

// a non-optimized way of checking for prime numbers:
bool do_check_prime(int x) // 为了体现效果, 该函数故意没有优化.
{
for (int i = 2; i < x; ++i)
if (x % i == 0)
return false;
return true;
}

int main()
{
// call function asynchronously:
std::future < bool > fut = std::async(do_check_prime, 194232491);

std::cout << "Checking...\n";
std::chrono::milliseconds span(1000); // 设置超时间隔.

// 如果超时，则输出"."，继续等待
while (fut.wait_for(span) == std::future_status::timeout)
std::cout << '.';

std::cout << "\n194232491 ";
if (fut.get()) // guaranteed to be ready (and not block) after wait returns
std::cout << "is prime.\n";
else
std::cout << "is not prime.\n";

return 0;
}

template <class Rep, class Period>
future_status wait_until (const chrono::time_point<Clock,Duration>& abs_time) const;

std::shared_future 介绍

shared_future() noexcept;

shared_future (const shared_future& x);

shared_future (shared_future&& x) noexcept;

shared_future (future<T>&& x) noexcept;

std::future_error 介绍

class future_error : public logic_error;

其他与 std::future 相关的函数介绍

template <class Fn, class... Args>
future<typename result_of<Fn(Args...)>::type>
async(Fn&& fn, Args&&... args);

template <class Fn, class... Args>
future<typename result_of<Fn(Args...)>::type>
async(launch policy, Fn&& fn, Args&&... args);

#include <stdio.h>
#include <stdlib.h>

#include <cmath>
#include <chrono>
#include <future>
#include <iostream>

double ThreadTask(int n) {
std::cout << std::this_thread::get_id()
<< " start computing..." << std::endl;

double ret = 0;
for (int i = 0; i <= n; i++) {
ret += std::sin(i);
}

std::cout << std::this_thread::get_id()
<< " finished computing..." << std::endl;
return ret;
}

int main(int argc, const char *argv[])
{
std::future<double> f(std::async(std::launch::async, ThreadTask, 100000000));

#if 0
while(f.wait_until(std::chrono::system_clock::now() + std::chrono::seconds(1))
!= std::future_status::ready) {
std::cout << "task is running...\n";
}
#else
while(f.wait_for(std::chrono::seconds(1))
!= std::future_status::ready) {
std::cout << "task is running...\n";
}
#endif

std::cout << f.get() << std::endl;

return EXIT_SUCCESS;
}

其他与 std::future 相关的枚举类介绍

enum class future_errc;

enum class future_status;

enum class launch;

取值

0

1

2

3

取值

0

1

2

#include <iostream>                // std::cout
#include <future>                // std::async, std::future, std::launch
#include <chrono>                // std::chrono::milliseconds
#include <thread>                // std::this_thread::sleep_for

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();
bar.get();
std::cout << '\n';

return 0;
}

with launch::async:
*@**@**@**@**@*@@@@@

with launch::deferred:
**********@@@@@@@@@@