现代C++之任务打包与多线程处理--使用std::packaged_task

朴灿烈づ我的快乐病毒、 2022-03-12 06:00 120阅读 0赞

解决的问题：  
多个互不相干的任务单线程处理较耗时，多线程将多任务均分后执行提高程序执行速度(类似线程池，本文更偏重于多任务的分配)。

--------------------

## 任务封装 ##

假设有任务类Task如下，每个任务初始化时传入要执行的参数，本例以传入n测试，任务执行结果以n\*n表示，具体执行函数在exec()函数中执行，假设每个任务耗时100毫秒：

class Task
    {
    public:
        Task(int n)
            :m_id(n)
            , m_result(0)
        {}
    
        int getId() { return m_id; }
        int getResult() { return m_result; }
        // 任务执行
        void exec()
        {
            // 假设每个任务需要100毫秒
            this_thread::sleep_for(std::chrono::milliseconds(100));
            cout << m_id << " ";
            // 测试假设将n*n作为每个任务的执行结果
            m_result = m_id * m_id;
        }
    
    private:
        int m_id;   // task id
        int m_result; // 任务执行结果
    };

--------------------

## 任务管理类 ##

TaskManager是一个任务管理类，它负责执行自己手头上的任务，构造函数中传入`vector<Task>`作为它的任务集。  
重载()执行手头上的所有任务，并将任务结果以std::map<int, int>形式返回。 根据线程个数分配同等个数的管理类对象，每个线程执行各自的任务管理类对象。

class TaskManager
    {
    public:
        TaskManager(const vector<Task>& tasks)
            : m_tasks(tasks) {}
    
        // 重载()
        std::map<int, int> operator()()
        {
            std::map<int, int> results{};
            for (auto& iter : m_tasks)
            {
                iter.exec();
                results[iter.getId()] = iter.getResult();
            }
    
            return results;
        };
    
    private:
        vector<Task> m_tasks{};
    };

--------------------

## 单线程测试 ##

// 假设有iAllTaskNumbers个任务 
        static const unsigned int iAllTaskNumbers = 50;
        // 总任务列表
        std::vector<Task> taskList{};
        for (int i = 0; i < iAllTaskNumbers; ++i)
        {
            taskList.emplace_back(Task(i));
        }
        
            // 单线程执行测试
    	auto start = std::chrono::system_clock::now();
     	cout << "single thread start" << endl;
    
    	TaskManager taskManager(taskList);  // 共iAllTaskNumbers个任务
    	auto result = taskManager();        // 单线程执行任务
    
    	cout << endl;
    	auto end = std::chrono::system_clock::now();
    	cout << "single thread finish" << endl;
    	auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
    	cout << "totaltime used : " << double(duration.count()) * std::chrono::microseconds::period::num / std::chrono::microseconds::period::den << " seconds." << endl;

测试50个任务执行耗时时间如下：  
![在这里插入图片描述][20190305130445163.png]

--------------------

## 多线程测试 ##

// 初步猜测推荐线程个数是4个
        static const unsigned int threadNumberGuess = 4;
        // 假设有iAllTaskNumbers个任务 
        static const unsigned int iAllTaskNumbers = 50;
        // 总任务列表
        std::vector<Task> taskList{};
        for (int i = 0; i < iAllTaskNumbers; ++i)
        {
            taskList.emplace_back(Task(i));
        }
        {
            // 多线程执行测试
    
            // 推荐线程个数hw
            unsigned int hw = std::thread::hardware_concurrency();
            // 实际开辟线程个数
            unsigned int realThreadNumbers = (hw != 0) ? hw : 4;
            cout << "thread numbers : " << realThreadNumbers << endl;
    
            // 任务管理对象作为函数对象
            std::vector<TaskManager> taskManagers{};
            for (unsigned int i = 0; i < realThreadNumbers; ++i)
            {
                // 根据线程个数将任务均分
                int begin = (i * taskList.size()) / realThreadNumbers;
                int end = (i + 1)*taskList.size() / realThreadNumbers;
    
                std::vector<Task> partTaskList{};
                for (auto j = begin; j < end; j++)
                {
                    partTaskList.emplace_back(taskList[j]);
                }
                taskManagers.emplace_back(partTaskList);
            }
    
            // 任务打包
            std::deque<std::packaged_task<std::map<int, int>()>> packagedTasks;
            for (unsigned int i = 0; i < realThreadNumbers; ++i)
            {
                std::packaged_task<map<int, int>()> task(taskManagers[i]);
                packagedTasks.emplace_back(move(task));
            }
    
            // 获取任务执行的返回值列表
            vector< std::future<std::map<int, int>>> runFutures{};
            for (unsigned int i = 0; i < realThreadNumbers; ++i)
            {
                runFutures.push_back(packagedTasks[i].get_future());
            }
    
            {
                auto start = std::chrono::system_clock::now();
                cout << "mutil thread start" << endl;
    
                // 每个任务在独立的线程中执行
                while (!packagedTasks.empty())
                {
                    std::packaged_task<std::map<int, int>()> myTask = std::move(packagedTasks.front());
                    packagedTasks.pop_front();
                    std::thread t(std::move(myTask));
                    t.detach();
                }
    
                // 获取执行结果,等待任务执行结束
                deque<map<int, int>> res{};
                for (unsigned int i = 0; i < realThreadNumbers; ++i)
                {
                    res.emplace_back(runFutures[i].get());
                }
    
                cout << endl;
                auto end = std::chrono::system_clock::now();
                cout << "mutil thread finish" << endl;
                auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
                cout << "totaltime used : " << double(duration.count()) * std::chrono::microseconds::period::num / std::chrono::microseconds::period::den << " seconds." << endl;
            }
        }

执行结果：  
![在这里插入图片描述][20190305130944133.png]  
4个线程，执行速度接近是单线程执行的4倍。

--------------------

## 完整程序源码 ##

[https://github.com/lesliefish/something/blob/master/packaged\_task\_test.cpp][https_github.com_lesliefish_something_blob_master_packaged_task_test.cpp]

[20190305130445163.png]: /images/20220312/7b85bd61b04445988f8d4a1e4dfb8618.png
[20190305130944133.png]: /images/20220312/2ea1500a57d64362a57e1dc88d5b15b1.png
[https_github.com_lesliefish_something_blob_master_packaged_task_test.cpp]: https://github.com/lesliefish/something/blob/master/packaged_task_test.cpp