C++的線程管理

線程類（Thread）
- 線程構造器
- - 約定構造器
  - 初始化構造器
  - 復制構造器
  - 移動構造器
多線程
- atomic
- condition_variable
- - 應用實列
- future
- - promise
  - - 應用實列
  - future
  - - 應用實列

線程類（Thread）

執行線程是一個指令序列，它可以在多線程環境中，與其他此類指令序列同時執行，同時共享相同的地址空間。

一個初始化的線程對象代表一個活動的執行線程；這樣的線程對象是可連接的，并且具有唯一的線程ID。

默認構造的（未初始化的）線程對象是不可連接的，并且它的線程 id 對于所有不可連接的線程都是通用的。

如果從可連接線程移出，或者對它們調用 join 或 detach，則該線程將變得不可連接。

#include <iostream>
#include <thread>
#include <unistd.h>using namespace std;void foo() 
{sleep(10);  //  sleep 10 secondscout << "I'm foo, awake now" << endl;
}void bar(int x)
{sleep(5);  //  sleep 10 secondscout << "I'm bar, awake now" << endl;
}int main() 
{thread T1 (foo);     // spawn new thread that calls foo()thread T2 (bar,0);  // spawn new thread that calls bar(0)cout << "main, foo and bar now execute concurrently...\n";// synchronize threads:T1.join();                // pauses until first finishesT2.join();               // pauses until second finishescout << "foo and bar completed.\n";return 0;
}

程序運行屏幕輸出

main, foo and bar now execute concurrently...
I'm bar, awake now
I'm foo, awake now
foo and bar completed.

線程構造器

- thread() noexcept;- template <class Fn, class... Args>explicit thread (Fn&& fn, Args&&... args); - thread (const thread&) = delete; - thread (thread&& x) noexcept;

約定構造器

thread() noexcept;

構造一個線程對象，它不包含任何執行線程。

初始化構造器

template <class Fn, class... Args>explicit thread (Fn&& fn, Args&&... args)；

構造一個線程對象，它擁有一個可連接執行線程。
新的執行線程調用 fn，并傳遞 args 作為參數。
此構造的完成與 fn 的副本開始運行同步。

#include <chrono>
#include <iostream>
#include <thread>
#include <utility>using namespace std;void f1(int n)
{for (int i = 0; i < 5; ++i){cout << "Thread 1 executing\n";++n;this_thread::sleep_for(chrono::milliseconds(10));}
}void f2(int& n, int sz)
{for (int i = 0; i < sz; ++i){cout << "Thread 2 executing\n";++n;this_thread::sleep_for(chrono::milliseconds(10));}
}int main()
{int n = 0;thread t2(f1, n + 1); 		 // pass by valuethread t3(f2, ref(n), 6); // pass by referencet2.join();t3.join();cout << "Final value of n is " << n << '\n';
}

程序運行屏幕輸出

Thread 1 executing
Thread 2 executing
Thread 1 executing
Thread 2 executing
Thread 2 executing
Thread 1 executing
Thread 1 executing
Thread 2 executing
Thread 1 executing
Thread 2 executing
Thread 2 executing
Final value of n is 6

復制構造器

thread (const thread&) = delete;

刪除構造函數，線程對象不能復制。

移動構造器

thread (thread&& x) noexcept;

構造一個線程對象，該對象獲取 x 表示的執行線程（如果有）。此操作不會以任何方式影響移動線程的執行，它只是傳輸其處理程序。
x 對象不再代表任何執行線程。

#include <chrono>
#include <iostream>
#include <thread>
#include <utility>
#include <unistd.h>using namespace std;void f2(int& n)
{thread::id this_id = this_thread::get_id();cout << "Thread " << this_id << " executing" << endl;for (int i = 0; i < 5; ++i){++n;this_thread::sleep_for(std::chrono::milliseconds(10));}
}int main()
{int n = 0;thread t3(f2, ref(n));thread t4(move(t3));t4.join();cout << "Final value of n is " << n << '\n';
}

程序運行屏幕輸出

Thread 140291256411904 executing
Final value of n is 5

多線程

atomic

atomic類型是封裝值的類型，保證其訪問不會導致數據爭用，并且可用于同步不同線程之間的內存訪問。

#include <iostream>
#include <atomic>
#include <thread>
#include <vector>
#include <random>using namespace std;atomic<bool> ready (false);
atomic_flag winner = ATOMIC_FLAG_INIT;void count1m (int id) {random_device dev;mt19937 rng(dev());uniform_int_distribution<mt19937::result_type> dist6(50,100); // distribution in range [1, 6]while (!ready) { this_thread::yield(); }int val = dist6(rng);	this_thread::sleep_for(chrono::milliseconds(val));if (!winner.test_and_set()) { cout << "thread #" << id << " won!\n"; }
}int main ()
{vector<thread> threads;cout << "5 threads compete...\n";for (int i=1; i<=5; ++i) threads.push_back(thread(count1m,i));ready = true;for (auto& th : threads) th.join();return 0;
}

程序運行2次，屏幕輸出

threads$ ./atomic
5 threads compete...
thread #3 won!
threads$ ./atomic
5 threads compete...
thread #4 won!

condition_variable

條件變量是一個能夠阻塞調用線程，直到通知恢復的對象。

當調用其等待函數之一時，它使用 unique_lock（通過互斥鎖）來鎖定線程。該線程將保持阻塞狀態，直到被另一個對同一 condition_variable 對象調用通知函數的線程喚醒。

Condition_variable 類型的對象始終使用 unique_lock 進行等待。

應用實列

#include <iostream>
#include <thread>
#include <mutex>
#include <vector>
#include <condition_variable>using namespace std;mutex mtx;
condition_variable cv;
bool ready = false;void wait_init_ready (int id) {unique_lock<mutex> lck(mtx);cout << "Init " << id << " start ..." << endl;while (!ready) cv.wait(lck);cout << "Init " << id << " done !!!" << '\n';
}void init_complete() {unique_lock<mutex> lck(mtx);ready = true;cv.notify_all();
}int main ()
{vector<thread> threads;for (int i=0; i<5; ++i)threads.push_back(thread(wait_init_ready, i));init_complete();for (auto& th : threads) th.join();return 0;
}

程序運行屏幕輸出

Init 0 start ...
Init 1 start ...
Init 3 start ...
Init 3 done !!!
Init 1 done !!!
Init 4 start ...
Init 4 done !!!
Init 0 done !!!
Init 2 start ...
Init 2 done !!!

future

具有允許異步訪問特定提供程序（可能在不同線程中）設置的值的功能的標頭。

這些提供者中的每一個（要么是promise或packaged_task對象，要么是對async的調用）與未來對象共享對共享狀態的訪問：提供者使共享狀態準備就緒的點與未來對象訪問共享狀態的點同步狀態。

promise

Promise 是一個對象，它可以存儲類型 T 的值，以便將來的對象（可能在另一個線程中）檢索，從而提供同步點。

在構造時，Promise 對象與一個新的共享狀態相關聯，它們可以在該狀態上存儲類型 T 的值或從 std::exception 派生的異常。

通過調用成員 get_future，可以將該共享狀態關聯到未來對象。調用后，兩個對象共享相同的共享狀態：

Promise 對象是異步提供者，預計會在某個時刻為共享狀態設置一個值。
future 對象是一個異步返回對象，可以檢索共享狀態的值，并在必要時等待它準備好。

共享狀態的生命周期至少持續到與其關聯的最后一個對象釋放它或被銷毀為止。因此，如果也與 future 相關聯，它可以在最初獲得它的 Promise 對象中存活下來。

應用實列

#include <iostream>
#include <functional>
#include <thread>
#include <future>using namespace std;struct data_pkt {int  id;uint8_t  data[20];
};void wait_new_value (future<data_pkt>& fut) {data_pkt x = fut.get();cout << "value: " << x.id << '\n';
}int main ()
{data_pkt  pkt;promise<data_pkt> prom;                      // create promisefuture<data_pkt> fut = prom.get_future();    // engagement with futurethread th1 (wait_new_value, ref(fut));  		// send future to new threadpkt.id = 1;prom.set_value (pkt);                        // fulfill promise// (synchronizes with getting the future)th1.join();return 0;
}

程序運行屏幕輸出

value: 1

future

future 是一個可以從某些提供程序對象或函數檢索值的對象，如果在不同的線程中，則可以正確同步此訪問。

“有效”未來是與共享狀態關聯的未來對象，并通過調用以下函數之一來構造：
異步
承諾::get_future
打包任務::獲取未來

future 對象僅在有效時才有用。默認構造的未來對象無效（除非移動分配了有效的未來）。

在有效的 future 上調用 future::get 會阻塞線程，直到提供者使共享狀態準備就緒（通過設置值或異常）。這樣，兩個線程可以通過一個等待另一個線程設置值來同步。

共享狀態的生命周期至少持續到與其關聯的最后一個對象釋放它或被銷毀為止。因此，如果與未來相關聯，共享狀態可以在最初獲取它的對象（如果有）中繼續存在。

應用實列

#include <iostream>
#include <future>
#include <chrono>
#include <signal.h>using namespace std;bool ready = false;
mutex mtx;
condition_variable cv;struct data_pkt {int   code;uint8_t  data[32];
};void term(int signum)
{if (signum == SIGINT){   printf("Received SIGINT(ctrl+c), exiting ... \n");unique_lock<mutex> lck(mtx);ready = true;cv.notify_all();}else{time_t mytime = time(0);printf("%d: %s\n", signum, asctime(localtime(&mytime)));printf("%d\n",signum);}
}bool async_promise (data_pkt &pkt) {cout << "async_promise start ..." << endl;struct sigaction act;act.sa_handler = term;sigaction(SIGQUIT, &act, NULL);sigaction(SIGINT, &act, NULL);unique_lock<mutex> lck(mtx);  while (!ready) cv.wait(lck);cout << "async_promise condition variable ready" << endl;	pkt.code = 1900;return true;
}int main ()
{data_pkt  pkt;// call function asynchronously:future<bool> fut = async (async_promise, ref(pkt)); // do something while waiting for function to set future:cout << "checking, please wait";chrono::milliseconds span (100);while (fut.wait_for(span) == future_status::timeout)cout << '.' << flush;bool x = fut.get();     // retrieve return valuecout << pkt.code << endl;return 0;
}

checking, please waitasync_promise start ...
............................^CReceived SIGINT(ctrl+c), exiting ...
async_promise condition variable ready
1900