C++多线程的用法（包含线程池小项目）

一些小tips: 

编译命令如下：

 g++ 7.thread_pool.cpp -lpthread

查看运行时间：

time ./a.out

 获得本进程的进程id：

this_thread::get_id()

 需要引入的库函数有：

#include<thread> // 引入线程库
#include<mutex> // 加入锁机制需要引入库函数mutex
#include<condition_variable> // 引入信号量机制

定义信号量、锁：

condition_variable m_cond
std::mutex m_mutex;

所谓的多线程只不过就是指定的某一个函数为入口函数，的另外一套执行流程。

什么是临界资源？多线程情况下，大家都能访问到的资源。

进程是资源分配的最基本单位，线程是进程中的概念。

线程也是操作系统分配的一批资源。一个线程的栈所占用的空间有多大？——8M。查看命令：

ulimit -a

简单用法：

#include<iostream>
#include<thread>
using namespace std;

#define BEGINS(x) namespace x{
#define ENDS(x) }

BEGINS(thread_usage)
void func() {
    cout << "hello wolrd" << endl;
    return ;
}
int main() {
    thread t1(func);  // t1已经开始运行了
    t1.join();        // 等待t1线程结束
    return 0;
}
ENDS(thread_usage)

int main() {
    thread_usage::main();

    return 0;
}

那么如何给入口函数传参呢？ 在C++中就很简单：

void print(int a, int b) {
    cout << a << " " << b << endl;
    return ;
}
int main() {
    thread t2(print, 3, 4);
    t2.join();
    return 0;
}

多线程封装思维：在相应的功能函数内部，不要去访问全局变量，类似于一个原子操作的功能。本身就支持多线程并行。

多线程程序设计：线程功能函数和入口函数要分开。（高内聚低耦合）

能不用锁就不用锁，因为这个锁机制会给程序运行效率带来极大的影响。

实现素数统计的功能

不加锁版：

#include<iostream>
#include<thread>
#include<cmath>
#include<mutex>
using namespace std;

#define BEGINS(x) namespace x{
#define ENDS(x) }

BEGINS(is_prime)
bool is_prime(int x) {
    for (int i = 2, I = sqrt(x); i <= I; i++) {
        if (x % i == 0) return false;
    }
    return true;
}
// 多线程——功能函数 
int prime_count(int l, int r) {
    // 从l到r范围内素数的数量
    int ans = 0;
    for (int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}
// 多线程——入口函数 
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin" << endl;
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "dnoe" << endl;
}
int main() {
    #define batch 500000
    thread *t[10];
    int ret[10];
    for (int i = 0, j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(worker, j, j + batch - 1, ref(ret[i]));
    }
    for (auto x : t) x->join();
    int ans = 0;
    for (auto x : ret) ans += x;
    for (auto x : t) delete x;
    cout << ans << endl;
    #undef batch
    return 0;
}
ENDS(is_prime)

int main() {
    // thread_usage::main();
    is_prime::main();
    return 0;
}

加锁版：

#include<iostream>
#include<thread>
#include<cmath>
#include<mutex>
using namespace std;

#define BEGINS(x) namespace x{
#define ENDS(x) }
BEGINS(prime_count2) 
int ans = 0;
std::mutex m_mutex;
bool is_prime(int x) {
    for (int i = 2, I = sqrt(x); i <= I; i++) {
        if (x % i == 0) return false;
    }
    return true;
}
void prime_count(int l, int r) {
    cout << this_thread::get_id() << " begin\n";
    for (int i = l; i <= r; i++) {
        std::unique_lock<std::mutex> lock(m_mutex); // 临界区
        ans += is_prime(i);
        lock.unlock();    
    }
    cout << this_thread::get_id() << " done\n";
    return ;
}
int main() {
    #define batch 500000
    thread *t[10];
    for (int i = 0, j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(prime_count, j, j + batch - 1);
    }
    for (auto x : t) x->join();
    for (auto x : t) delete x;
    cout << ans << endl;
    #undef batch
    return 0;
}
ENDS(prime_count2)

int main() {
    prime_count2::main();
    return 0;
}

为什么不用++而是用+=

因为后者是原子操作，而前者不是，在多线程情况下可能存在覆盖写的问题。

__sync_fetch_and_add

这个函数也是原子操作。

#include<iostream>
#include<thread>
#include<cmath>
#include<mutex>
using namespace std;

#define BEGINS(x) namespace x{
#define ENDS(x) }

BEGINS(thread3)
int ans = 0;
bool is_prime(int x) {
    for (int i = 2, I = sqrt(x); i <= I; i++) {
        if (x % i == 0) return false;
    }
    return true;
}
void prime_count(int l, int r) {
    cout << this_thread::get_id() << "begin\n";
    for (int i = l; i <= r; i++) {
        int ret = is_prime(i);
        __sync_fetch_and_add(&ans, ret);
    }
    cout << this_thread::get_id() << "done\n";
}
int main() {
    #define batch 500000
    thread *t[10];
    for (int i = 0, j = 1; i < 10; i++, j += batch) {
        t[i] = new thread(prime_count, j, j + batch - 1);
    }
    for (auto x : t) x->join();
    for (auto x : t) delete x;
    cout << ans << endl;
    #undef batch 
    return 0;
}
ENDS(thread3)

int main() {
    thread3::main();
    return 0;
}

线程池的实现

我们希望申请内存的动作是可控的。线程作为一种内存资源，在通常的设计模式下，申请线程资源并不可控。

解决办法就是线程池。在线程池内部，线程数量是可控的。把计算任务打包，扔到一个任务队列中。

什么是计算任务？——分为过程（函数方法）和数据（函数参数），如何进行打包——bind()方法。

线程池解决了传统的多线程程序设计中，面对不同任务我们需要实现不同程序的麻烦：而是直接往里塞任务就行了。这样就实现了资源的有效利用和管控。

m_cond.notify_one() 对于这个进程来说，是否同时也解锁了互斥资源？

是的，m_cond.notify_one()会解锁互斥资源。在调用m_cond.notify_one()之前，通常会先调用m_mutex.lock()来锁定互斥资源，然后在适当的时候调用m_cond.notify_one()来唤醒等待该条件的线程。在唤醒线程后，该线程会重新尝试获取互斥资源的锁，而其他等待的线程会继续等待。所以，m_cond.notify_one()不仅唤醒等待的线程，还会解锁互斥资源，使得等待的线程有机会获取互斥资源的锁。

子进程的中wait函数对互斥量进行解锁，同时线程进入阻塞或者等待状态。

/*************************************************************************
        > File Name: threadpool.cpp
        > Author: jby
        > Mail: 
        > Created Time: Wed 13 Sep 2023 08:48:23 AM CST
 ************************************************************************/

#include<iostream>
#include<cmath>
#include<vector>
#include<unordered_map>
#include<queue>
#include<mutex>
#include<thread>
#include<condition_variable>
#include<functional>
using namespace std;

#define BEGINS(x) namespace x {
#define ENDS(x) }

BEGINS(thread_pool_test)
class Task {
public:
    template<typename FUNC_T, typename ...ARGS>
    Task(FUNC_T func, ARGS ...args) {
        this->func = bind(func, forward<ARGS>(args)...);
    }
    void run() {
        func();
        return ;
    }
private:
    function<void()> func; // 任意函数
};
class ThreadPool {
public:
    ThreadPool(int n = 1) : thread_size(n), threads(n), starting(false) {
        this->start();
        return ;
    }
    void worker() {
        auto id = this_thread::get_id(); // 获得本进程号
        running[id] = true;
        while (running[id]) {
            // 取任务
            Task *t = get_task(); 
            t->run();
            delete t;
        }
        return ;
    }
    void start() {
        if (starting == true) return ; // 如果已经开始了就不用启动了
        for (int i = 0; i < thread_size; i++) {
            threads[i] = new thread(&ThreadPool::worker, this);
        }
        starting = true;
        return ;
    }
    template<typename FUNC_T, typename ...ARGS>
    void add_task(FUNC_T func, ARGS ...args) {
        unique_lock<mutex> lock(m_mutex); 
        tasks.push(new Task(func, forward<ARGS>(args)...)); // 任务池相当于临界资源
        m_cond.notify_one(); // 生产者消费者模型
        return ;
    }
    void stop() {
        if (starting == false) return ; // 如果已经关了就不用再关了
        for (int i = 0; i < threads.size(); i++) { // 往队列末尾投递毒药任务
            this->add_task(&ThreadPool::stop_runnnig, this);
        }
        for (int i = 0; i < threads.size(); i++) {
            threads[i]->join();
        }
        for (int i = 0; i < threads.size(); i++) {
            delete threads[i]; // 释放那片进程剩余的空间
            threads[i] = nullptr; // 进程指针指向空
        }
        starting = false;
        return ;
    }
    ~ThreadPool() {
        this->stop();
        while (!tasks.empty()) { // 如果任务队列里还有任务没执行完，全部丢弃
            delete tasks.front();
            tasks.pop();
        }
        return ;
    }
private:
    bool starting;
    int thread_size;
    Task *get_task() {
        unique_lock<mutex> lock(m_mutex);
        while (tasks.empty()) m_cond.wait(lock); // 生产者消费者模型 //子进程的中wait函数对互斥量进行解锁，同时线程进入阻塞或者等待状态。
        Task *t = tasks.front();
        tasks.pop();
        return t;
    }
    std::mutex m_mutex;
    std::condition_variable m_cond;
    vector<thread *> threads; // 线程池子
    unordered_map<decltype(this_thread::get_id()), bool> running; // 进程号到运行状态的哈希映射
    queue<Task *> tasks;  // 任务队列
    void stop_runnnig() { // 毒药任务
        auto id = this_thread::get_id();
        running[id] = false;
        return ;
    }
};
bool is_prime(int x) {
    for (int i = 2, I = sqrt(x); i <= I; i++) {
        if (x % i == 0) return false;
    }
    return true;
}
int prime_count(int l, int r) {
    int ans = 0;
    for (int i = l; i <= r; i++) {
        ans += is_prime(i);
    }
    return ans;
}
void worker(int l, int r, int &ret) {
    cout << this_thread::get_id() << "begin\n";
    ret = prime_count(l, r);
    cout << this_thread::get_id() << "done\n";
    return ;
}
int main() {
    #define batch 500000
    ThreadPool tp(5); // 五个任务的窗口队列
    int ret[10];
    for (int i = 0, j = 1; i < 10; i++, j += batch) {
        tp.add_task(worker, j, j + batch - 1, ref(ret[i]));
    }
    tp.stop();
    int ans = 0;
    for (auto x : ret) ans += x;
    cout << ans << endl;
    #undef batch
    return 0;
}
ENDS(thread_pool_test)

int main() {
    thread_pool_test::main();
    return 0;
}