一個簡易的綫程池

duqn · 發表於 2024-3-23 20:26:53

ThreadPool.h:

// ThreadPool.h
#ifndef THREADPOOL_H
#define THREADPOOL_H
#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <future>
#include <functional>
#include <stdexcept>
class ThreadPool {
public:
ThreadPool(size_t);
~ThreadPool();
// 声明模板成员函数
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;
};
#include "ThreadPool.tpp" // 包含模板实现
#endif // THREADPOOL_H

複製代碼

ThreadPool.tpp

// ThreadPool.tpp
#ifndef THREADPOOL_TPP
#define THREADPOOL_TPP
#include "ThreadPool.h"
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);
// don't allow enqueueing after stopping the pool
if(stop)
throw std::runtime_error("enqueue on stopped ThreadPool");
tasks.emplace([task](){ (*task)(); });
}
condition.notify_one();
return res;
}
#endif // THREADPOOL_TPP

複製代碼

ThreadPool.cpp

// pool.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(auto &worker: workers)
worker.join();
}

複製代碼

main.cpp

#include <iostream>
#include <string>
#include <random>
#include <chrono>
#include <unordered_map>
#include <mutex>
#include "ThreadPool.h"
std::string generateRandomString(size_t length) {
const std::string chars = "abcdefghijklmnopqrstuvwxyz";
std::random_device random_device;
std::mt19937 generator(random_device());
std::uniform_int_distribution<> distribution(0, chars.size() - 1);
std::string random_string;
for (size_t i = 0; i < length; ++i) {
random_string += chars[distribution(generator)];
}
return random_string;
}
void calculateCharacterFrequencies(const std::string& input, int threadNum) {
auto start = std::chrono::high_resolution_clock::now();
std::unordered_map<char, int> frequencies;
for (char c : input) {
frequencies[c]++;
}
auto end = std::chrono::high_resolution_clock::now();
std::chrono::duration<double, std::milli> duration = end - start;
// Thread-safe output
static std::mutex cout_mutex;
{
std::lock_guard<std::mutex> lock(cout_mutex);
std::cout << input << ", Thread" << threadNum << ", ";
for (const auto& pair : frequencies) {
std::cout << pair.first << ":" << pair.second << ",";
}
std::cout << "time:" << duration.count() << "ms\n";
}
}
int main() {
ThreadPool pool(5);
std::vector<std::future<void>> futures;
auto totalStart = std::chrono::high_resolution_clock::now();
for (int i = 0; i < 100; ++i) {
size_t length = 30 + rand() % 71; // Random length between 30 and 100
std::string randomString = generateRandomString(length);
futures.push_back(pool.enqueue(calculateCharacterFrequencies, randomString, i % 5 + 1));
}
// Wait for all tasks to complete
for (auto& fut : futures) {
fut.get();
}
auto totalEnd = std::chrono::high_resolution_clock::now();
std::chrono::duration<double, std::milli> totalDuration = totalEnd - totalStart;
std::cout << "Total execution time: " << totalDuration.count() << "ms\n";
return 0;
}

複製代碼

編譯命令：

g++ -std=c++11 -pthread ThreadPool.cpp main.cpp -o main

複製代碼

duqn · 發表於 2024-3-23 20:28:10

这段代码实现了以下功能：

generateRandomString(size_t length)函数生成一个指定长度的随机字符串，仅包含小写字母。
calculateCharacterFrequencies(const std::string& input, int threadNum)函数计算一个字符串中每个字符出现的次数，并打印出来，包括处理该任务的线程号和任务执行所需的时间。这里使用了一个static std::mutex来确保输出时的线程安全。
在main函数中，我们首先生成了100个随机字符串，并将它们的处理任务提交到了线程池中。我们通过enqueue方法的返回值（一个std::future对象）来追踪任务的完成情况。完成所有任务后，计算并输出总体执行时间。

duqn · 發表於 2024-3-23 20:32:59

一個簡單的調用示例

#include <iostream>
#include <string>
#include <chrono>
#include "ThreadPool.h"
// 一个简单的数学计算函数
int multiply(int a, int b) {
return a * b;
}
// 一个字符串操作函数
std::string concat(const std::string& a, const std::string& b) {
return a + b;
}
// 一个模拟耗时操作的函数
std::string simulateLongComputationTask(const std::string& input) {
std::this_thread::sleep_for(std::chrono::seconds(3)); // 模拟耗时任务
return "Processed: " + input;
}
int main() {
ThreadPool pool(4); // 创建一个拥有4个工作线程的线程池
// 任务1: 数学计算
auto task1 = pool.enqueue(multiply, 5, 6);
// 任务2: 字符串操作
auto task2 = pool.enqueue(concat, "Hello, ", "World!");
// 任务3: 耗时任务
auto task3 = pool.enqueue(simulateLongComputationTask, "Task 3");
// 获取并输出结果
std::cout << "Task 1 result: " << task1.get() << std::endl;
std::cout << "Task 2 result: " << task2.get() << std::endl;
std::cout << "Task 3 result: " << task3.get() << std::endl;
return 0;
}

複製代碼