【高性能计算】C++多线程计算与线程池

线程

• 一个进程中可以并发多个线程，每条线程并行执行不同的任务。
• 因为虚拟内存的关系不同的进程互相不知道对方的存在，相对独立，互相有独立的内存空间，地址空间；
• 相同进程内的线程之间共享进程的内存空间，文件描述符表，部分寄存器等资源，但是每个线程都有自己独立的线程栈和部分寄存器，线程间访问资源需要考虑互斥问题。
• 进程创建，销毁，切换代价大；
• 线程创建，销毁，切换小。
• 进程有利于资源管理和保护，开销大；
• 线程不利于资源管理和保护，需要使用锁保证线程的安全运行，开销小
• 不同进程间的线程通信等同于进程通信，一般是五种通信方式：
  消息队列；
  共享内存；
  有名管道/无名管道；
  信号；
  scoket。
• 相同进程中的不同线程因为资源共享不存在通信问题，主要是资源的互斥访问，一般需要资源加锁以防死锁。

线程池

• 线程池采用预创建的技术，在应用程序启动之后，将立即创建一定数量的线程(N1)，放入空闲队列中。这些线程都是处于阻塞（Suspended）状态，不消耗CPU，但占用较小的内存空间。
• 当任务到来后，缓冲池选择一个空闲线程，把任务传入此线程中运行。当N1个线程都在处理任务后，缓冲池自动创建一定数量的新线程，用于处理更多的任务。
• 在任务执行完毕后线程也不退出，而是继续保持在池中等待下一次的任务。
• 当系统比较空闲时，大部分线程都一直处于暂停状态，线程池自动销毁一部分线程，回收系统资源。
• 线程池的主要组成部分有二：
  (1)任务队列（Task Queue）
  (2)线程池（Thread Pool）
• 线程池通常适合下面的几个场合：
  (1)单位时间内处理任务频繁而且任务处理时间短
  (2)对实时性要求较高。如果接受到任务后在创建线程，可能满足不了实时要求，因此必须采用线程池进行预创建。
  一个分享，这里，代码非常的简洁，只有一个头文件ThreadPool.h，这里贴出来作为备份。

#ifndef THREAD_POOL_H
#define THREAD_POOL_H
 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
 
class ThreadPool {
public:
    ThreadPool(size_t);
    template
    auto enqueue(F&& f, Args&&... args) 
        -> std::future::type>;
    ~ThreadPool();
private:
    // need to keep track of threads so we can join them
    std::vector< std::thread > workers;
    // the task queue
    std::queue< std::function > tasks;
    
    // synchronization
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};
 
// the constructor just launches some amount of workers
inline ThreadPool::ThreadPool(size_t threads)
    :   stop(false)
{
    for(size_t i = 0;i task;
 
                    {
                    	// 锁定互斥锁以确保没有其他人正在访问该资源
                        std::unique_lock 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();
                }
            }
        );
}
 
// add new work item to the pool
template
auto ThreadPool::enqueue(F&& f, Args&&... args) 
    -> std::future::type>
{
    using return_type = typename std::result_of::type;
 	// 创建一个执行
    auto task = std::make_shared< std::packaged_task >(
            std::bind(std::forward(f), std::forward(args)...)
        );
        
    std::future res = task->get_future();
    {
        std::unique_lock 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;
}
 
// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for(std::thread &worker: workers)
        worker.join();
}
 
#endif
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

#include 
#include "ThreadPool.h"
 
void func()
{
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    std::cout<<"worker thread ID:"<