std::packaged_task簡介

std::packaged_task是C++11開始加入到STL(Standard Template Libarary, 標準模板庫)的併發程式設計工具,位於標頭檔案 <future>中, 使用時只需要包含該標頭檔案即可

使用場景

假設我們的可執行程式有如下要求: 某些第三方提供的API未使用加鎖保護機制,但這些API呼叫卻不允許在不同執行緒裡併發的呼叫, 比如涉及到單埠通訊的API以及操作UI控制元件的API, 同一時間段內僅能進行一次呼叫, 這種情況下, 就必須要保證該系列的API的呼叫必須在單執行緒中呼叫. 具體怎麼做呢?

應對方案

方案1: 使用任務佇列

開啟獨立執行緒, 將所有涉及到相關API呼叫的操作放入某個序列的任務佇列中, 然後在開啟的獨立執行緒中迴圈呼叫

方案2: 使用事件迴圈

在某些流程中, 可能會使用到事件迴圈, 並提供了相應的事件觸發介面將對應的事件加入到主事件迴圈佇列中, 由事件響應者進行任務呼叫

方案分析

對於方案1, 我們的工作量相對比較少, 建立任務佇列, 建立執行緒並迴圈執行佇列中的任務 對於方案2, 我們的步驟相對繁瑣一些, 針對該系列任務定製事件, 定義事件響應回撥

綜合分析

以上方案均能保證我們的API可以合理的放到某單個執行緒(主執行緒或新執行緒)中去, 但是一般情況下, 通過這種方式, 我們只能將任務放到單執行緒去執行, 而如果相獲取結果的話, 在C++11之前還是相當麻煩的.不過自從C++11, 我們有了一個很厲害的神兵利器用來解決我們的這一需求:

發起非同步任務, 等待執行結果

具體實現

1. 將所有相關的API呼叫任務做一層通用的包裝, 比如放到std::function<void()>中

2. 使用現有的事件邏輯或者開啟新的執行緒去執行對應包裝的任務

3. 使用std::packaged_task將任務和任務引數進行任務包裝, 可以設定不同的返回值, 根據具體api的情況而定, 比如:

    int do_something(int arg),		可以包裝為std::packaged_task<int()>)  
   

4. 獲取std::packaged_task的future, 用來獲取包裝起來的任務中的返回值

auto
 
 future = task.get_future();

5. 將 包裝好的task通過已有的事件觸發介面或者自定義的推送到任務佇列的介面將task放入std::function中推送到任務佇列中
6. 呼叫future.wait() 或者future.get()等方式來等待返回值, 如果不惜要返回值, 也可以直接wait()即可, 這樣既保證了api的呼叫一定是在單執行緒, 並且發起呼叫的執行緒中還可以使用同步的方式等待結果

具體示例程式碼

#include <queue>
#include <future>
#include <functional>
#include <iostream>
#include <thread>


int main(int, char **)
{
    int ticks = 0;
    /*訪問佇列時的保護鎖*/
    std::mutex task_lock;

	/*這是全域性的任務佇列*/
    std::queue<std::function<void()>> task_list;

	/*後臺發起任務的呼叫*/
    std::thread background([&]{


        while(ticks < 100) {
            auto task = std::make_shared<std::packaged_task<bool()>>(std::bind([=] (int ticks) -> bool {
                std::cout << "[Main] execute task ticks: " << ticks << std::endl;
                return true;
            }, ticks));
            auto future = task->get_future();
            std::cout << "[Background] get task future" << std::endl;
            {
                std::lock_guard<std::mutex> lock(task_lock);
                std::cout << "[Background] push task, ticks: " << ticks << std::endl;
                task_list.push([=] {
                   if(task) (*task)();
                });
            }
            std::this_thread::sleep_for(std::chrono::seconds(1));
            ticks++;
        }
    });

	/*主執行緒迴圈執行佇列任務*/
    std::queue<std::function<void()>> tasks;
    while (ticks < 100) {
        if(task_list.empty()) {
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        } else {
            {
                std::lock_guard<std::mutex> lock(task_lock);
                tasks.swap(task_list);
            }
            while(!tasks.empty()) {
                auto task = tasks.front();
                if(task) {
                    task ();
                }
                tasks.pop();
            }
        }
    }

    background.join();

    return 0;
}