C++ 多线程使用

2024-12-10 本文已影响0人 leon0514

线程

std::thread

创建线程

使用函数

#include <thread>
#include <iostream>

void task(int i)
{
    std::cout << i << std::endl;
}

int main()
{
    for (int i = 0; i < 8; i++)
    {
        std::thread t(task, i);
        t.detach();
    }
    getchar();
    return 0;
}

使用lambda表达式创建线程

#include <thread>
#include <iostream>

int main()
{
    for (int i = 0; i < 8; i++)
    {
        std::thread t([](int i){
            std::cout << i << std::endl;
        }, i);
        t.detach();
    }
    getchar();
    return 0;
}

重载了()运算符的类的实例

#include <thread>
#include <iostream>

struct Task
{
    void operator()(int i)
    {
        std::cout << i << std::endl;
    }
};

int main()
{
    for (int i = 0; i < 8; i++)
    {
        Task task;
        std::thread t(task, i);
        t.detach();
    }
    getchar();
    return 0;
}

在类中创建线程

#include <iostream>
#include <thread>
#include <atomic>
#include <chrono>

class Worker {
public:
    Worker() : stopFlag(false) {}

    // 启动线程
    void start() {
        workerThread = std::thread(&Worker::run, this);
    }

    // 停止线程
    void stop() {
        stopFlag = true;
        if (workerThread.joinable()) {
            workerThread.join();
        }
    }

    // 析构函数确保线程停止
    ~Worker() {
        stop();
    }

private:
    std::thread workerThread;
    std::atomic<bool> stopFlag;

    // 线程运行的函数
    void run() {
        while (!stopFlag) {
            std::cout << "线程正在运行..." << std::endl;
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
        std::cout << "线程已停止" << std::endl;
    }
};

int main() {
    Worker worker;

    worker.start(); // 启动线程

    std::this_thread::sleep_for(std::chrono::seconds(5)); // 主线程等待

    worker.stop(); // 停止线程

    return 0;
}

参数传递

值传递

#include <thread>
#include <iostream>

struct Task
{
    void operator()(int i)
    {
        printf("thread &i : %p\n", &i);
    }
};

int main()
{
    for (int i = 0; i < 1; i++)
    {
        printf("&i : %p\n", &i);
        Task task;
        std::thread t(task, i);
        t.detach();
    }
    getchar();
    return 0;
}

引用传递
引用传递的地址和主线程的地址也是不一致的

#include <thread>
#include <iostream>

struct Task
{
    void operator()(const int &i) // 这里必须是const，不然会报错
    {
        printf("thread &i : %p\n", &i);
    }
};

int main()
{
    for (int i = 0; i < 1; i++)
    {
        printf("&i : %p\n", &i);
        Task task;
        std::thread t(task, i);
        t.detach();
    }
    getchar();
    return 0;
}

指针传递

#include <thread>
#include <iostream>

struct Task
{
    void operator()(int *i)
    {
        printf("thread &i : %p\n", i);
    }
};

int main()
{
    for (int i = 0; i < 1; i++)
    {
        printf("&i : %p\n", &i);
        Task task;
        std::thread t(task, &i);
        t.join(); // 这里使用join，确保i没有被销毁
    }
    getchar();
    return 0;
}

std::ref 传参
std::ref 可以保证子线程中的参数地址和主线程中的参数地址是一致的，即在子线程和主线程是同步的。

#include <thread>
#include <iostream>

struct Task
{
    void operator()(int &i)
    {
        printf("thread &i : %p\n", &i);
    }
};

int main()
{
    for (int i = 0; i < 1; i++)
    {
        printf("&i : %p\n", &i);
        Task task;
        std::thread t(task, std::ref(i));
        t.join();
    }
    getchar();
    return 0;
}

锁

std::mutex

构造函数
lock函数调用线程将锁住该互斥量。
unlock函数调用线程将释放该互斥量。

try_lock函数调用线程将尝试锁住该互斥量，不成功返回false。

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex

volatile int counter(0); // non-atomic counter
std::mutex mtx;           // locks access to counter

void attempt_10k_increases() 
{
    for (int i=0; i<10000; ++i) 
    {
        mtx.lock();
        ++counter;
        mtx.unlock();
    }
}

int main (int argc, const char* argv[]) {
    std::thread threads[10];
    for (int i=0; i<10; ++i)
        threads[i] = std::thread(attempt_10k_increases);

    for (auto& th : threads) th.join();
    std::cout << counter << " successful increases of the counter.\n";

    return 0;
}

std::lock_guard

自动锁定和解锁
当 std::lock_guard 对象被创建时，它会自动锁定提供的互斥锁。当该对象离开其作用域（即被销毁）时，它会自动解锁互斥锁。这种自动管理确保了即使在发生异常的情况下，互斥锁也能被正确解锁。
简单易用
std::lock_guard 的使用非常简单，只需要在需要保护的代码块之前创建它即可。无需显式调用锁定和解锁函数。

非递归
std::lock_guard 不支持递归锁定，即不能多次锁定同一个互斥锁。如果尝试这样做，会导致未定义的行为。

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex

volatile int counter(0); // non-atomic counter
std::mutex mtx;           // locks access to counter

void attempt_10k_increases() 
{
    for (int i=0; i<10000; ++i) 
    {
        std::lock_guard<std::mutex> l(mtx);
        ++counter;
    }
}

int main (int argc, const char* argv[]) {
    std::thread threads[10];
    for (int i=0; i<10; ++i)
        threads[i] = std::thread(attempt_10k_increases);

    for (auto& th : threads) th.join();
    std::cout << counter << " successful increases of the counter.\n";

    return 0;
}

std::unique_lock

可以在构造时选择是否锁定互斥量。std::unique_lock<std::mutex> lock(my_mutex1, std::defer_lock); std::defer_lock的意思就是并没有给mutex加锁：初始化一个没有加锁的mutex，但是后面需要对unique_lock对象进行加锁
支持手动锁定和解锁：可以在需要时调用 lock() 和 unlock() 方法。

支持条件变量的等待：在等待条件变量时，可以传入 unique_lock，并在条件满足时自动解锁和重新锁定。

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex

volatile int counter(0); // non-atomic counter
std::mutex mtx;           // locks access to counter

void attempt_10k_increases() 
{
    for (int i=0; i<10000; ++i) 
    {
        std::unique_lock<std::mutex> l(mtx);
        ++counter;
    }
}

int main (int argc, const char* argv[]) {
    std::thread threads[10];
    for (int i=0; i<10; ++i)
        threads[i] = std::thread(attempt_10k_increases);

    for (auto& th : threads) th.join();
    std::cout << counter << " successful increases of the counter.\n";

    return 0;
}

条件变量

std::condition_variable

等待

条件变量的wait方法通常是通过以下方式实现的：

把线程放入等待队列并释放锁。
等待被显式唤醒（notify_one 或 notify_all）或者因为虚假唤醒被操作系统强行唤醒。即使没有执行notify_one或者notify_all函数也有可能部分线程被虚假唤醒。
被唤醒后重新尝试获取锁。

void wait(std::unique_lock<std::mutex>& lock);

template <class Predicate>  
void wait (unique_lock<mutex>& lck, Predicate pred); // 解决虚假唤醒

唤醒

void notify_one() noexcept;
void notify_all() noexcept;

DEMO

#include <iostream>       // std::cout
#include <thread>         // std::thread
#include <mutex>          // std::mutex


std::mutex mtx;
std::condition_variable cv;
bool ready = false;   // 全局标志位

void printId(int id)
{
    std::unique_lock<std::mutex> lck(mtx);
    // 如果标志位不为true，则等待
    while(!ready)
    {
        // 线程被阻塞，直到标志位变为true，此时 mtx 被释放，go 线程能够获取到锁。
        cv.wait(lck);
    }
    // 另一种写法, 等价于上面的写法
    // cv.wait(lck, []{ return ready; }); 
    std::cout << "thread: " << std::this_thread::get_id() << " id: " << id << "\n";
}

void go()
{
    std::unique_lock<std::mutex> lck(mtx);
    // 改变全局标志位
    ready = true;
    // 唤醒所有线程
    cv.notify_all();
}

int main()
{
    std::thread threads[10];

    for (int i = 0; i < 10; ++i)
    {
        threads[i] = std::thread(printId, i);
    }
    std::cout << "create done.\n" ;

    go();

    for (auto &t : threads)
    {
        t.join();
    }
    std::cout << "process done.\n" ;
    return 0;
}

数据传递

promise future

承诺的未来
std::promise的作用就是提供一个不同线程之间的数据同步机制，它可以存储一个某种类型的值，并将其传递给对应的future，即使这个future不在同一个线程中也可以安全的访问到这个值。

/** @file  20190815future.cpp
*  @note  
*  @brief
*  @author 
*  @date   2019-8-15
*  @note   
*  @history
*  @warning
*/
#include <iostream>
#include <functional>
#include <future>
#include <thread>
#include <chrono>
#include <cstdlib>

void thread_set_promise(std::promise<int>& promiseObj) {
    std::cout << "In a thread, making data...\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    promiseObj.set_value(35);
    std::cout << "Finished\n";
}

int main() {
    std::promise<int> promiseObj;
    std::future<int> futureObj = promiseObj.get_future();
    std::thread t(&thread_set_promise, std::ref(promiseObj));
    std::cout << futureObj.get() << std::endl;
    t.join();

    system("pause");
    return 0;
}

生产者消费者

c++多线程实现生产者消费者

线程池

c++11 线程池