在MFC中实现一个简单的线程池

[墨香年少 32]

线程池的头文件 ThreadPool.h

#pragma once

#include <queue>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>

class ThreadPool {
public:
    ThreadPool(size_t numThreads);
    ~ThreadPool();

    template<class F, class... Args>
    void enqueue(F&& f, Args&&... args);

private:
    // Need to keep track of threads so we can join them
    std::vector<std::thread> workers;

    // The task queue
    std::queue<std::function<void()>> tasks;

    // Synchronization
    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop;
};

// Constructor
inline ThreadPool::ThreadPool(size_t numThreads) : stop(false) {
    for (size_t i = 0; i < numThreads; ++i) {
        workers.emplace_back(
            [this] {
                for (;;) {
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(this->queueMutex);
                        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();
                }
            }
        );
    }
}

// Destructor
inline ThreadPool::~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        stop = true;
    }

    condition.notify_all();
    for (std::thread& worker : workers) {
        worker.join();
    }
}

// Add new work item to the pool
template<class F, class... Args>
void ThreadPool::enqueue(F&& f, Args&&... args) {
    {
        std::function<void()> task = std::bind(std::forward<F>(f), std::forward<Args>(args)...);

        std::unique_lock<std::mutex> lock(queueMutex);
        tasks.emplace(std::move(task));
    }

    condition.notify_one();
}

假设我们有一个需要使用线程池执行的复杂计算任务，比如计算一个大型矩阵的乘法。以下是一个简单的示例：

Example.cpp:

#include "ThreadPool.h"
#include <iostream>
#include <vector>

// Function to multiply two matrices
void multiplyMatrices(const std::vector<std::vector<int>>& matrix1, const std::vector<std::vector<int>>& matrix2, std::vector<std::vector<int>>& result, size_t startRow, size_t endRow) {
    size_t n = matrix1.size();
    size_t m = matrix2[0].size();
    size_t p = matrix2.size();

    for (size_t i = startRow; i < endRow; ++i) {
        for (size_t j = 0; j < m; ++j) {
            result[i][j] = 0;
            for (size_t k = 0; k < p; ++k) {
                result[i][j] += matrix1[i][k] * matrix2[k][j];
            }
        }
    }
}

int main() {
    const size_t numThreads = 4;
    ThreadPool pool(numThreads);

    // Example matrices
    std::vector<std::vector<int>> matrix1 = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
    std::vector<std::vector<int>> matrix2 = {{9, 8, 7}, {6, 5, 4}, {3, 2, 1}};
    std::vector<std::vector<int>> result(matrix1.size(), std::vector<int>(matrix2[0].size(), 0));

    // Define the number of rows each thread should handle
    size_t rowsPerThread = matrix1.size() / numThreads;

    // Enqueue tasks for each thread
    for (size_t i = 0; i < numThreads; ++i) {
        size_t startRow = i * rowsPerThread;
        size_t endRow = (i == numThreads - 1) ? matrix1.size() : startRow + rowsPerThread;
        pool.enqueue(multiplyMatrices, std::ref(matrix1), std::ref(matrix2), std::ref(result), startRow, endRow);
    }

    // Wait for all tasks to complete
    // (In this example, we're not synchronizing explicitly because ThreadPool's destructor handles it)
    
    // Print the result
    std::cout << "Resultant matrix:\n";
    for (const auto& row : result) {
        for (int num : row) {
            std::cout << num << " ";
        }
        std::cout << "\n";
    }

    return 0;
}

ThreadPool 类: 线程池类管理一组线程，并提供了任务队列和同步机制来安全地执行任务。

enqueue 方法: 用于将任务添加到队列中，可以接受任意的函数和参数，并在需要时执行。

ComplexTaskExample.cpp: 展示了如何使用线程池来并行计算两个矩阵的乘法。每个线程负责处理矩阵的一部分行，以提高计算效率。