C++11 線程並發
並發 頭文件<future> <thread>
高級接口
async()、future<>
future<int> result1; //int為func1返回值
result1 = async(func1); //啟動func1,但有可能被推遲,直到調用get或wait
或
future<int> result1(async(func1));
result1.get();//獲得返回值
async()的參數可以是函數、成員函數、函數對象或lambda
async([]{})
//絕不推遲
future<int> result1 = async(lunch::async, func1);
//強制延緩,直到調用f.get()或f.wait()
future<int> f(async(lunch::deferred, func1));
auto f1 = sync(lunch::deferred, task1);
auto f2 = sync(lunch::deferred, task2);
auto val = b?f1.get():f2.get();
auto f1 = async(task1);
try
{
f1.get();
}
catch (exception* e)
{
cerr << "EXCEPTION: " << e->what() << endl;
}
一個future<>只能被調用get()一次,之後future就處於無效狀態,
這種狀態可調用valid()來檢測,
只要對future調用wait(),就可以強制啟動該future象征的線程並等待這一後臺操作終止
future<...> f(asybc(func));
...
f.wait();
f.wait_for(seconds(10));
f.wait_until(system_clock::now()+chrono::minutes(1));
不論wait_for()或wait_until()都可以返回一下三種東西之一:
1.future_status::deferred
如果async()延緩了操作而程序中又完全沒有調用wait()或get()
2.future_status::timeout
如果某個操作被異步啟動但尚未結束,而waiting又已逾期
3.future_status::ready
操作已完成
future<...> f(async(task));
如果在單線程環境,有可能延緩而沒有啟動
所以先檢查
if (f.wait_for(chrono::seconds(0)) != future_status::deferred){
while (f.wait_for(chrono::seconds(0)) != future_status::ready)){
...
this_thread::yeild();
//或this_thread::sleep_for(milliseconds(100));
}
}
...
auto r = f.get();
例子:
#include <chrono>
#include <iostream>
#include <future>
#include <thread>
#include <random>
#include <exception>
using namespace std;
using namespace std::chrono;
int doSomething(char c)
{
default_random_engine dre(c);
uniform_int_distribution<int> id(10, 1000);
for (int i = 0; i < 10; ++i)
{
this_thread::sleep_for(milliseconds(id(dre)));
cout.put(c).flush();
}
return c;
}
int main()
{
auto f1 = async([]{doSomething(‘.‘); });
auto f2 = async([]{doSomething(‘+‘); });
if (f1.wait_for(seconds(0)) != future_status::deferred ||
f2.wait_for(seconds(0)) != future_status::deferred)
{
while (f1.wait_for(seconds(0)) != future_status::ready &&//有一個線程完成就跳出循環
f2.wait_for(seconds(0)) != future_status::ready)
{
this_thread::yield();//提示重新安排到下一個線程
}
}
cout.put(‘\n‘).flush();
try {
f1.get();
f2.get();
}
catch (const exception& e){
cout << "\nException: " << e.what() << endl;
}
cout << "\ndone" << endl;
}
傳遞實參
上例使用了lambda並讓它調用後臺函數
auto f1 = async([]{doSomething(‘.‘); });
也可以傳遞async語句之前就存在的實參
char c = ‘@‘;
auto f1 = async([=]{doSomething(c); });
[=]表示傳遞給lambda的是c的拷貝
char c = ‘@‘;
//傳遞c的拷貝
auto f1 = async([=]{doSomething(c); });
auto f1 = async(doSomething, c);
//傳遞c的引用
auto f1 = async([&]{doSomething(c); });
auto f1 = async(doSomething, ref(c));
如果使用async,就應該以值傳遞方式來傳遞所有用來處理目標函數的必要object,
使async只需使用局部拷貝
調用成員函數
class X
{
public:
void mem(int num){ cout << "memfunc: " << num << endl; };
};
int main()
{
X x;
auto f = async(&X::mem, x, 42); //x.mem(42);
f.get();
}
傳給async一個成員函數的pointer,之後的第一個實參必須是個reference或pointer,指向某個object
多次處理
shared_future 可多次調用get()
int queryNum()
{
cout << "read number: ";
int num;
cin >> num;
if (!cin)
{
throw runtime_error("no number read");
}
return num;
}
void doSomething1(char c, shared_future<int> f)
{
try{
int num = f.get();
for (int i = 0; i < num; ++i)
{
this_thread::sleep_for(chrono::milliseconds(100));
cout.put(c).flush();
}
}
catch (const exception& e)
{
cerr << "EXCEPTION in thread " << this_thread::get_id() << ": " << e.what() << endl;
}
}
int main()
{
try{
shared_future<int> f = async(queryNum);
auto f1 = async(launch::async, doSomething1, ‘.‘, f);
auto f2 = async(launch::async, doSomething1, ‘+‘, f);
auto f3 = async(launch::async, doSomething1, ‘*‘, f);
f1.get();
f2.get();
f3.get();
}
catch (const exception& e){
cout << "\nEXCEPTION: " << e.what() << endl;
}
cout << "\ndone" << endl;
}
確保f的壽命不短於被啟動的線程
低層接口
Thread 和 Promise
std::thread
void doSomething();
thread t(doSomething); //啟動
...
t.join();//等待結束
t.detach();//卸載
cout << thread::hardware_concurrency() << endl; //CPU線程個數
和async相比
1.沒有所謂發射策略,試著將目標函數啟動與一個新線程中,如果無法做到會拋出system_error,並帶有錯誤碼resource_unavailable_try_again
2.沒有接口處理線程結果,唯一可獲得的是一個獨一無二的線程ID
3.如果發生異常,但未捕捉於線程之內,程序會立刻中止並調用terminate()
4.你必須聲明是否想要等待線程結束(join)
或打算將它自母體卸載使它運行與後臺而不受任何控制(detach)
如果你在thread object壽命結束前不這麽做,或如果它發生了一次move assignment,
程序會中止並調用terminate()
4.如果你讓線程運行於後臺而main結束了,所有線程會被魯莽而硬性地終止
Detached Thread(卸離後的線程)
一般性規則:
Detached Thread應該寧可只訪問local copy
。。。
Thread ID
this_thread::get_id()
thread t(doSomething2, 5, ‘.‘);
t.get_id()
thread::id(); //默認構造,生成一個獨一無二的ID用來表現 no thread
Promise
void doSomething3(promise<string>& p)
{
try{
cout << "read char (‘x‘ for exception): ";
char c = cin.get();
if (c == ‘x‘){
throw runtime_error(string("char ") + c + " read");
}
string s = string("char ") + c + " processed";
p.set_value(move(s));//存放一個值
//p.set_value_at_thread_exit(move(s));//線程結束時存放
}
catch (...){
p.set_exception(current_exception());//存放一個異常
//p.set_exception_at_thread_exit(current_exception());
}
}
int main()
{
try {
promise<string> p;
thread t(doSomething3, ref(p));//傳入一個promise引用
t.detach();
future<string> f(p.get_future());//get_future取出值
cout << "result: " << f.get() << endl;//取得存儲結果//get會停滯,直到p.set_value
}
catch (const exception& e){
cerr << "EXCEPTION: " << e.what() << endl;
}
catch (...){
cerr << "EXCEPTION " << endl;
}
}
this_thread
this_thread::get_id()
this_thread::sleep_for(duration)
this_thread::sleep_until(timepoint)
this_thread::yeild() //建議釋放控制以便重新調度,讓下一個線程能夠執行
當心 Concurrency (並發)
多個線程並發處理相同數據而又不曾同步化,那麽唯一安全的情況就是:所有線程只讀取數據
除非另有說明,c++標準庫提供的函數通常不支持讀或寫動作與另一個寫動作(寫至同一筆數據)並發執行
Mutex 和 Lock
mutex
mutex valMutex;
valMutex.lock();
++val;
valMutex.unlock();
lock_guard:
{
//lock、析構函數自動釋放lock(unlock) (加一個大括號使其更快釋放)
lock_guard<mutex> lg(valMutex);
++val;
}
//recursive_mutex 允許同一線程多次鎖定,並在最近一次(last)相應的unlock時釋放lock
recursive_mutex dbMutex;
lock_guard<recursive_mutex> lg(dbMutex);
嘗試性的Lock
mutex m;
//試圖獲得一個lock,成功返回true,為了仍能夠使用lock_guard,要額外傳入實參 adopt_lock
//有可能假性失敗,lock為被他人拿走也可能失敗返回false
while (m.try_lock() == false)
{
//doSomeOtherStuff();
}
lock_guard<mutex> lg1(m, adopt_lock);
timed_mutex m1; //recursive_timed_mutex
//if (m1.try_lock_until(system_clock::now() + chrono::seconds(1)))
if (m1.try_lock_for(chrono::seconds(1))){
lock_guard<timed_mutex> lg2(m1, adopt_lock);
}
else
{
//couldNotGetTheLock();
}
處理多個Lock,易發生互鎖,死鎖
使用全局函數lock解決
mutex
mutex m1;
mutex m2;
{
//lock會阻塞,直到所有mutex被鎖定或出現異常
lock(m1, m2);
//成功鎖定後使用lock_guard,以adopt_lock作為第二實參,確保這些mutex在離開作用域時解鎖
lock_guard<mutex> lockM2(m1, adopt_lock);
lock_guard<mutex> lockM3(m2, adopt_lock);
//...
}//自動unlock
//或者使用try_lock,取得所有lock時返回-1,否則返回第一失敗的lock的索引且其它成功的會unlock
int idx = try_lock(m1, m2);
if (idx < 0)
{
lock_guard<mutex> lockM2(m1, adopt_lock);
lock_guard<mutex> lockM3(m2, adopt_lock);
//...
}//自動unlock
else
{
cerr << "could not lock mutex m" << idx + 1 << endl;
}
//使用lock、try_lock後使用adopt_lock過繼給lock_guard才能自動解鎖
只調用一次
once_flag oc;
call_once(oc, initialize);//保證initialize函數只調用一次
once_flag oc1;
call_once(oc1, []{staticData = initializeStaticData(); });
class X
{
private:
mutable once_flag initDataFlag;
void initData()const;
public:
int GetData()const{
call_once(initDataFlag, &X::initData, this);
//...
}
};
Condition Variable(條件變量) <condition_variable>
future從某線程傳遞數據到另一線程只能一次,且future的主要目的是處理線程的返回值或異常
條件變量可用來同步化線程之間的數據流邏輯依賴關系
使用ready flag(一個bool變量)讓某線程等待另一線程是一個粗淺辦法
while (!readyFlag){
//...
this_thread::yeild();
}
消耗寶貴的CPU時間重復檢查flag,
此外很難找出適當的sleep周期,
2次檢查間隔太短則仍舊浪費CPU時間於檢查動作上,太長則會發生延誤
一個較好的做法是使用條件變量,使一個線程可以喚醒一或多個其他等待中的線程
包含<mutex> <condition_variable>
mutex readyMutex;
condition_variable readyCondVar;
//激發的條件終於滿足的線程(或多線程之一)必須調用
readyCondVar.notify_one();
//或
readyCondVar.notify_all();
//等待條件滿足的線程必須調用
unique_lock<mutex> l(readyMutex);
readyCondVar.wait(l);
但是有可能出現假醒,即wait在condition varibale尚未被notified時便返回
假醒無法被預測,實質上是隨機的
在wakeup之後你仍需要代碼去驗證條件實際已達成,
例如必須檢查數據是否真正備妥,或仍需要諸如ready flag之類的東西
#include <condition_variable>
#include <mutex>
#include <future>
#include <iostream>
using namespace std;
bool readyFlag; //表示條件真的滿足了
mutex readyMutex;
condition_variable readyCondVar;
void thread1()
{
cout << "<return>" << endl;
cin.get();
{
lock_guard<mutex> lg(readyMutex);
readyFlag = true;
}
readyCondVar.notify_one();
}
void thread2()
{
{//這裏必須使用unque_lock,不可使用lock_guard,因為wait的內部會明確地對mutex進行解鎖和鎖定
unique_lock<mutex> ul(readyMutex);
//lambda當做第二實參,用來檢查條件是否真的滿足,直到返回true
readyCondVar.wait(ul, []{return readyFlag; });
//相當於
//while (!readyFlag){
// readyCondVar.wait(ul);
//}
}
cout << "done" << endl;
}
int main()
{
auto f1 = async(launch::async, thread1);
auto f2 = async(launch::async, thread2);
f1.wait();
f2.wait();
}
例子
std::queue<int> queue1;//並發使用,被一個mutex和一個condition variable保護著
mutex queueMutex;
condition_variable queueCondVar;
void provider(int val)
{
for (int i = 0; i < 6; ++i)
{
{
lock_guard<mutex> ul(queueMutex);
queue1.push(val + i);
}
queueCondVar.notify_one();
this_thread::sleep_for(chrono::milliseconds(val));
}
}
void consumer(int num)
{
while (true)
{
int val;
{
unique_lock<mutex> ul(queueMutex);
queueCondVar.wait(ul, []{return !queue1.empty(); });
val = queue1.front();
queue1.pop();
cout << "consumer" << num << ": " << val << endl;
//使用wait_for wait_until則要判斷返回值
//if (queueCondVar.wait_for(ul, chrono::seconds(1), []{return !queue1.empty(); }))
//if (queueCondVar.wait_for(ul, chrono::seconds(1)) == cv_status::no_timeout)//這個沒有判斷readyFlag
//{
// val = queue1.front();
// queue1.pop();
// cout << "consumer" << num << ": " << val << endl;
//}
}
}
}
int main()
{
auto p1 = async(launch::async, provider, 100);
auto p2 = async(launch::async, provider, 300);
auto p3 = async(launch::async, provider, 500);
auto c1 = async(launch::async, consumer, 1);
auto c2 = async(launch::async, consumer, 2);
//c1.wait();
getchar();
exit(0);
}
atomic
即使基本數據類型,讀寫也不是atomic(不可切割的),readyFlag可能讀到被寫一半的bool
編譯器生成的代碼有可能改變操作次序
借由mutex可解決上述2個問題,但從必要的資源和潛藏的獨占訪問來看,
mutex也許是個相對昂貴的操作,所以也許值得以atomic取代mutex和lock
atomic<bool> readyFlag(false);
void thread1()
{
//...
readyFlag.store(true);//賦予一個新值
}
void thread2()
{
while (!readyFlag.load())//取當前值
{
this_thread::sleep_for(chrono::milliseconds(100));
}
//...
}
void provider()
{
while (true)
{
cout << "<return>" << endl;
char arr[100];
cin.getline(arr, 100);
data = 42;
readyFlag.store(true);
}
}
void consumer()
{
while (true)
{
while (!readyFlag.load()){
cout.put(‘.‘).flush();
this_thread::sleep_for(chrono::milliseconds(50));
}
cout << "\nvalue : " << data << endl;
readyFlag = false;
}
}
int main()
{
auto p = async(launch::async, provider);
auto c = async(launch::async, consumer);
c.wait();
}
atomic<int> ai(0);
int x = ai;
ai = 10;
ai++;
ai -= 17;
atomic<int> a;
atomic_init(&a, 0);//沒有初始化則使用這個函數進行初始化
#include "stdafx.h" //#include <memory> #include <chrono> #include <ctime> #include <string> #include <iostream> #include <future> #include <thread> #include <random> #include <exception> #include <vector> using namespace std; using namespace std::chrono; int doSomething(/*const char&*/char c) { default_random_engine dre(c); uniform_int_distribution<int> id(10, 1000); for (int i = 0; i < 10; ++i) { this_thread::sleep_for(milliseconds(id(dre))); cout.put(c).flush(); } return c; } int func1() { return doSomething(‘.‘); } int func2() { return doSomething(‘+‘); } #include <list> void task1() { list<int> v; while (true) { for (int i = 0; i < 1000000; ++i) { v.push_back(i); } cout.put(‘.‘).flush(); } } #if 0 int quickComputation(); //快速直接 quick and dirty int accurateComputation(); //精確但是慢 future<int> f; int bestResultInTime() { auto tp = chrono::system_clock::now() + chrono::minutes(1); f = async(launch::async, accurateComputation); int guess = quickComputation(); future_status s = f.wait_until(tp); //等待 if (s == future_status::ready) //完成 { return f.get(); } else { return guess; } } #endif #if 0 int main() { //傳遞實參 //auto f1 = async([]{ doSomething(‘.‘); }); char c = ‘@‘; //傳遞c的拷貝 //auto f1 = async([=]{ doSomething(c); }); //auto f1 = async(doSomething, c); //傳遞c的引用 //auto f1 = async([&]{ doSomething(c); });//這個需要把doSomething(const char& c) ??? auto f1 = async( doSomething, ref(c)); auto f2 = async([]{ doSomething(‘+‘); }); c = ‘_‘; if (f1.wait_for(seconds(0)) != future_status::deferred || f2.wait_for(seconds(0)) != future_status::deferred) { while (f1.wait_for(seconds(0)) != future_status::ready && //有一個線程完成就跳出循環 f2.wait_for(seconds(0)) != future_status::ready) { this_thread::yield();//提示重新安排到下一個線程 } } cout.put(‘\n‘).flush(); try { f1.get(); //一個future只能調用一次get,之後future處於無效狀態 f2.get(); } catch (const exception& e){ cout << "\nException: " << e.what() << endl; } cout << "\ndone" << endl; //cout << f1._Get_value() << endl; //相當於 cout << f1._Is_ready() << endl; cout << f1.valid() << endl; } #endif // 0 #if 0 class X { public: void mem(int num){ cout << "memfunc: " << num << endl; }; }; int main() { X x; auto f = async(&X::mem, x, 42); //x.mem(42); f.get(); } #endif // 0 #if 0 int queryNum() { cout << "read number: "; int num; cin >> num; if (!cin) { throw runtime_error("no number read"); } return num; } void doSomething1(char c, shared_future<int> f) { try{ int num = f.get(); for (int i = 0; i < num; ++i) { this_thread::sleep_for(chrono::milliseconds(100)); cout.put(c).flush(); } } catch (const exception& e) { cerr << "EXCEPTION in thread " << this_thread::get_id() << ": " << e.what() << endl; } } int main() { try{ shared_future<int> f = async(queryNum); auto f1 = async(launch::async, doSomething1, ‘.‘, f); auto f2 = async(launch::async, doSomething1, ‘+‘, f); auto f3 = async(launch::async, doSomething1, ‘*‘, f); f1.get(); f2.get(); f3.get(); } catch (const exception& e){ cout << "\nEXCEPTION: " << e.what() << endl; } cout << "\ndone" << endl; } #endif // 0 #include <thread> #if 0 void doSomething2(int num, char c) { try { default_random_engine dre(42 * c); uniform_int_distribution<int> id(10, 1000); for (int i = 0; i < num; ++i) { this_thread::sleep_for(milliseconds(id(dre))); cout.put(c).flush(); } } catch (...){ cerr << "THREAD-EXCEPTION (thread " << this_thread::get_id() << ")" << endl; } } int main() { try { thread t1(doSomething2, 5, ‘.‘); cout << "- start fg thread " << t1.get_id() << endl; for (int i = 0; i < 5; ++i) { thread t(doSomething2, 10, ‘a‘ + i); cout << "-detach start bg thread " << t.get_id() << endl; t.detach(); } cin.get(); cout << "- join fg thread " << t1.get_id() << endl; t1.join(); } catch (const exception& e){ cerr << "EXCEPTION: " << e.what() << endl; } } #endif // 0 #if 0 void doSomething3(promise<string>& p) { try{ cout << "read char (‘x‘ for exception): "; char c = cin.get(); if (c == ‘x‘){ throw runtime_error(string("char ") + c + " read"); } string s = string("char ") + c + " processed"; p.set_value(move(s));//存放一個值 //p.set_value_at_thread_exit(move(s));//線程結束時存放 } catch (...){ p.set_exception(current_exception());//存放一個異常 //p.set_exception_at_thread_exit(current_exception()); } } int main() { try { promise<string> p; thread t(doSomething3, ref(p));//傳入一個promise引用 t.detach(); future<string> f(p.get_future());//get_future取出值 cout << "result: " << f.get() << endl;//取得存儲結果//get會停滯,直到p.set_value } catch (const exception& e){ cerr << "EXCEPTION: " << e.what() << endl; } catch (...){ cerr << "EXCEPTION " << endl; } } #endif // 0 mutex printMutex; void print(const string& s) { lock_guard<mutex> l(printMutex); for (char c : s) { cout.put(c); } cout << endl; } int initialize() { return 0; } vector<string> initializeStaticData() { vector<string> staticData; return staticData; } vector<string> staticData; #if 0 int main() { auto f1 = async(launch::async, print, "Hello from a first thread"); auto f2 = async(launch::async, print, "Hello from a second thread"); print("Hello from the main thread"); try {//確保mutex銷毀之前,f1,f2 線程完成 f1.wait(); f2.wait(); } catch (...) { } //遞歸的 Lock 造成死鎖,在第二次lock拋出異常system_error,錯誤碼resource_deadlock_would_occur //lock_guard<mutex> l(printMutex); //print("Hello from the main thread"); recursive_mutex dbMutex; lock_guard<recursive_mutex> lg(dbMutex); //recursive_mutex 允許同一線程多次鎖定,並在最近一次(last)相應的unlock時釋放lock mutex m; //試圖獲得一個lock,成功返回true,為了仍能夠使用lock_guard,要額外傳入實參 adopt_lock //有可能假性失敗,lock為被他人拿走也可能失敗返回false while (m.try_lock() == false) { //doSomeOtherStuff(); } lock_guard<mutex> lg1(m, adopt_lock); timed_mutex mm; //recursive_timed_mutex //if (m1.try_lock_until(system_clock::now() + chrono::seconds(1))) if (mm.try_lock_for(chrono::seconds(1))){ lock_guard<timed_mutex> lg2(mm, adopt_lock); } else { //couldNotGetTheLock(); } mutex m1; mutex m2; { //lock會阻塞,直到所有mutex被鎖定或出現異常 lock(m1, m2); //成功鎖定後使用lock_guard,以adopt_lock作為第二實參,確保這些mutex在離開作用域時解鎖 lock_guard<mutex> lockM2(m1, adopt_lock); lock_guard<mutex> lockM3(m2, adopt_lock); //... }//自動unlock //或者使用try_lock,取得所有lock時返回-1,否則返回第一失敗的lock的索引且其它成功的會unlock int idx = try_lock(m1, m2); if (idx < 0) { lock_guard<mutex> lockM2(m1, adopt_lock); lock_guard<mutex> lockM3(m2, adopt_lock); //... }//自動unlock else { cerr << "could not lock mutex m" << idx + 1 << endl; } //使用lock、try_lock後使用adopt_lock過繼給lock_guard才能自動解鎖 //只調用一次 once_flag oc; call_once(oc, initialize);//保證initialize函數只調用一次 once_flag oc1; call_once(oc1, []{staticData = initializeStaticData(); }); } class X { private: mutable once_flag initDataFlag; void initData()const; public: int GetData()const{ call_once(initDataFlag, &X::initData, this); //... } }; #endif // _DEBUG #if 0 //條件變量 #include <condition_variable> //mutex readyMutex; //condition_variable readyCondVar; ////激發的條件終於滿足的線程(或多線程之一)必須調用 //readyCondVar.notify_one(); //一個 ////或 //readyCondVar.notify_all(); //所有 ////等待條件滿足的線程必須調用 //unique_lock<mutex> l(readyMutex); //readyCondVar.wait(l); bool readyFlag; //表示條件真的滿足了 mutex readyMutex; condition_variable readyCondVar; void thread1() { cout << "<return>" << endl; cin.get(); { lock_guard<mutex> lg(readyMutex); readyFlag = true; } readyCondVar.notify_one(); } void thread2() { {//這裏必須使用unque_lock,不可使用lock_guard,因為wait的內部會明確地對mutex進行解鎖和鎖定 unique_lock<mutex> ul(readyMutex); //lambda當做第二實參,用來檢查條件是否真的滿足,直到返回true readyCondVar.wait(ul, []{return readyFlag; }); //相當於 //while (!readyFlag){ // readyCondVar.wait(ul); //} } cout << "done" << endl; } int main() { auto f1 = async(launch::async, thread1); auto f2 = async(launch::async, thread2); f1.wait(); f2.wait(); } #endif // 0 #include <queue> #if 0 std::queue<int> queue1;//並發使用,被一個mutex和一個condition variable保護著 mutex queueMutex; condition_variable queueCondVar; void provider(int val) { for (int i = 0; i < 6; ++i) { { lock_guard<mutex> ul(queueMutex); queue1.push(val + i); } queueCondVar.notify_one(); this_thread::sleep_for(chrono::milliseconds(val)); } } void consumer(int num) { while (true) { int val; { unique_lock<mutex> ul(queueMutex); queueCondVar.wait(ul, []{return !queue1.empty(); }); val = queue1.front(); queue1.pop(); cout << "consumer" << num << ": " << val << endl; //使用wait_for wait_until則要判斷返回值 //if (queueCondVar.wait_for(ul, chrono::seconds(1), []{return !queue1.empty(); })) //if (queueCondVar.wait_for(ul, chrono::seconds(1)) == cv_status::no_timeout)//這個沒有判斷readyFlag //{ // val = queue1.front(); // queue1.pop(); // cout << "consumer" << num << ": " << val << endl; //} } } } int main() { auto p1 = async(launch::async, provider, 100); auto p2 = async(launch::async, provider, 300); auto p3 = async(launch::async, provider, 500); auto c1 = async(launch::async, consumer, 1); auto c2 = async(launch::async, consumer, 2); //c1.wait(); getchar(); exit(0); } #endif // 0 #include <atomic> void foo(){ atomic<int> ai(0); int x = ai; ai = 10; ai++; ai -= 17; atomic<int> a; atomic_init(&a, 0);//沒有初始化則使用這個進行初始化 } long data; atomic<bool> readyFlag(false); //void thread1() //{ // //... // readyFlag.store(true);//賦予一個新值 //} // //void thread2() //{ // while (!readyFlag.load())//取當前值 // { // this_thread::sleep_for(chrono::milliseconds(100)); // } // //... //} void provider() { while (true) { cout << "<return>" << endl; char arr[100]; cin.getline(arr, 100); data = 42; readyFlag.store(true); } } void consumer() { while (true) { while (!readyFlag.load()){ cout.put(‘.‘).flush(); this_thread::sleep_for(chrono::milliseconds(50)); } cout << "\nvalue : " << data << endl; readyFlag = false; } } int main() { auto p = async(launch::async, provider); auto c = async(launch::async, consumer); c.wait(); } //與條件變量一起使用 //unique_lock<mutex> ul(readyMutex); //readyCondVar.wait(ul, []{return !readyFlag.load(); });
C++11 線程並發