c++多线程输出数字相加

duqn · 發表於 2024-3-23 21:56:02

ThreadPool.h

#ifndef THREADPOOL_H
#define THREADPOOL_H
#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>
class ThreadPool {
public:
ThreadPool(size_t threads);
~ThreadPool();
// 在类定义内声明并定义enqueue方法
template<class F, class... Args>
auto enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type>;
private:
std::vector<std::thread> workers;
std::queue<std::function<void()>> tasks;
std::mutex queue_mutex;
std::condition_variable condition;
bool stop;
};
// 在头文件中定义enqueue，确保编译器能够找到模板定义
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type> {
using return_type = typename std::result_of<F(Args...)>::type;
auto task = std::make_shared< std::packaged_task<return_type()> >(
std::bind(std::forward<F>(f), std::forward<Args>(args)...)
);
std::future<return_type> res = task->get_future();
{
std::unique_lock<std::mutex> lock(queue_mutex);
if (stop)
throw std::runtime_error("enqueue on stopped ThreadPool");
tasks.emplace([task]() { (*task)(); });
}
condition.notify_one();
return res;
}
#endif

複製代碼

ThreadPool.cpp

#include "ThreadPool.h"
ThreadPool::ThreadPool(size_t threads)
: stop(false) {
for (size_t i = 0; i < threads; ++i)
workers.emplace_back(
[this] {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(this->queue_mutex);
this->condition.wait(lock,
[this] { return this->stop || !this->tasks.empty(); });
if (this->stop && this->tasks.empty())
return;
task = std::move(this->tasks.front());
this->tasks.pop();
}
task();
}
}
);
}
ThreadPool::~ThreadPool() {
{
std::unique_lock<std::mutex> lock(queue_mutex);
stop = true;
}
condition.notify_all();
for (std::thread& worker : workers)
worker.join();
}

複製代碼

main.cpp

#include <iostream>
#include <mutex>
#include <chrono>
#include <random>
#include "ThreadPool.h"
std::mutex cout_mutex; // 用于同步标准输出的互斥锁
int getRandomNumberBetween(int a, int b) {
// 静态用于性能考虑：初始化一次，避免每次调用都初始化
static std::random_device rd; // 用于获得种子
static std::mt19937 gen(rd()); // 标准 mersenne_twister_engine seeded with rd()
std::uniform_int_distribution<> distrib(a, b);
return distrib(gen);
}
void taskFunction(int id, int& number) {
auto start = std::chrono::high_resolution_clock::now();
std::default_random_engine generator(std::chrono::system_clock::now().time_since_epoch().count());
std::uniform_int_distribution<int> distribution(100, 300);
for (int i = 0; i < getRandomNumberBetween(20,70); ++i) {
++number;
std::this_thread::sleep_for(std::chrono::milliseconds(distribution(generator)));
}
auto end = std::chrono::high_resolution_clock::now();
std::chrono::duration<double> diff = end - start;
// 在任务内部进行输出
std::lock_guard<std::mutex> guard(cout_mutex); // 使用互斥锁来保护输出
std::cout << "thread" << id << ", used:" << diff.count() << "s, result:" << number << std::endl;
}
int main() {
ThreadPool pool(10); // 创建一个拥有10个线程的线程池
int numbers[10] = {}; // 定义10个数字
// 统计开始时间
auto start = std::chrono::high_resolution_clock::now();
std::vector<std::future<void>> futures;
for (int i = 0; i < 10; ++i) {
futures.push_back(pool.enqueue(taskFunction, i + 1, std::ref(numbers[i])));
}
// 等待所有任务完成
for (auto& future : futures) {
future.get();
}
// 统计结束时间
auto end = std::chrono::high_resolution_clock::now();
std::chrono::duration<double> diff = end - start;
std::cout << "all done! total use:" << diff.count() << "s" << std::endl;
// 注意：这里没有实现“最快的线程”和“最慢的线程”的功能，因为这需要在taskFunction中额外维护线程执行时间的信息，并在所有任务完成后进行比较。
// 可以通过增加一个结构体来保存每个线程的执行时间和ID，然后在所有任务完成后进行排序来实现。
return 0;
}

複製代碼

结果：

thread8, used:4.68329s, result:23
thread3, used:5.15709s, result:25
thread7, used:5.38293s, result:27
thread9, used:5.87792s, result:28
thread10, used:6.02598s, result:30
thread5, used:6.2226s, result:29
thread1, used:6.53804s, result:32
thread4, used:6.66759s, result:30
thread2, used:6.92785s, result:35
thread6, used:8.11339s, result:37
all done! total use:8.14531s

複製代碼