并发编程 02 管理线程

发布于 2021-10-04  119 次阅读


  在这一章,我们将先从基础开始学习如何启动线程、等待完成,或是在后台运行它。之后我们将进一步讨论线程函数启动时想起传递额外的参数,以及如何将线程的所有权从一个std::thread转移到另一个。最后我们会看一看选择所使用的线程数量,以及标识特定的线程。

一、基本线程管理

  每个C++程序至少有一个线程,它是由C++在运行时启动的,该线程运行着main()函数。你的程序可以继续启动其他函数作为线程入口。然后,这些线程连同初始线程一起,并发运行。正如程序会在main()返回时结束一样,当指定的函数返回时,该线程就会退出。下面我们来看一下如何启动一个线程。

1.1 启动线程

  我们可以通过一个函数来构造一个std::thread:

void func();
std::thread th(func);

  与许多C++标准库相似,std::thread可以与任何可调用(Callable)类型一起工作,所以我们可以传入一个函数对象的实例进去:

class Task
{
public:
    operator()() const
    {
        func_1();
        func_2();
    }
};

Task task;
std::thread(task);

  在这种情况下,所提供的函数对象被复制(Copied)到新创建的执行线程的存储器中,并在哪里调用。因此重要的是,副本与原本都有等效行为。
  但是我们有一件非常需要注意的事情——所谓的“C++的最棘手的解析”。如果你尝试传递一个匿名变量,那么其语法声明可能与函数声明一样,编译器可能误认为一个函数。下面我们来看看它:

std::thread th(Task());

  上述代码声明了一个函数th(),他接受单个参数(参数类型是指向不接受参数、同时返回Task对象的函数的指针),并返回std::thread对象。它是一个函数,而不是创建一个对象。
  现在我们有两种方法来避免这个问题。

std::thread th( (Task()) );
// 增加一层()将Task()括住
std::thread th{ Task() };
// 使用新的统一初始化语法

  同时我们也可以使用lambda表达式来完成(bind同理):

std::thread th([]
{
    func_1();
    func_2();
});

  一旦开始了线程,你需要显式地决定是要等待它完成(通过结合),还是让他自学运行(通过分离)。如果你在std::thread对象被销毁前未做决定,那么你的程序会被终止(std::thread的析构函数调用std::terminate())。因此,即便在异常存在的情况下,保证线程正确地结合或分离都是我们的当务之急。
  如果我们不等待线程完成,那么则需要保证该线程访问的数据是有效的,直到该线程完成。
  考虑下面一种情况:当线程函数持有局部变量的指针或者引用,且当函数退出的时候线程未完成。

struct Func
{
    int& i;
    Func(int& p_i) : i(p_i) {}
    operator()()
    {
        for (int j = 0; j < 10e3; ++j)
        {
            // do sth
        }
    }
};

void oops()
{
    int local_state = 0;
    Func func(local_state);
    std::thread th(func);
    th.detach();
    // 不等待线程完成而直接退出程序
}

  在上述情况下,如果我们不等待线程th完成操作,在退出oops()之后,func所访问的引用则是一个被销毁的变量。对于指针、引用的问题,我们可以使用两种方法解决:一是使用copy的方式传入变量;二是等待线程完成(结合线程,藉由th.join()实现)。对于第一种方法,在我们推出oops()函数时,线程th中的变量将被销毁。特别地,在一个访问局部变量的函数中创建线程是一个糟糕的主意,除非在退出函数前等待线程完成。

1.2 等待线程完成

  我们可以使用thread.join()来等待线程完成。但是在这种情况下,在独立的线程上运行函数是没有什么意义的,因为第一个线程在此期间(等待另一个线程完成)做不了任何有用的事情。在实际代码中,初始线程要么有自己的工作要去做,要么是在等待所有线程完成之前就要启动多个线程。
  join()很简单——要么等待一个线程完成、要么不等。如果需要对线程就行更细粒度的控制,比如线程检查是否完成,或只是在一段特定的时间内进行等待,那么就必须使用特定机制代替,比如条件变量和future,我们将在第4章提到。调用join()的行为也会清理所有与该线程有关的存储器,这样std::thread对象就不在与已完成的线程相关联,他也不与任何线程相关联。这就意味着,我们只能对一个线程调用一次join(),一旦调用了join(),该线程就不是可连接的,并且joinable()放回false。

1.3 在异常环境下等待

  如上所述,我们需要确保在std::thread对象被销毁前已调用join()或者detach()函数。如果要分离线程,在线程启动后就可以立即调用detach(),所以这不是重点。问题在于,如果我们需要等待线程,就需要仔细选择在哪个地方调用join()。这意味着,如果在线程启动后、join()调用前发生了异常,那么对join()的调用很容易被跳过。
  为了避免这种情况,我们可以使用异常捕获(try/catch),下面演示使用标准的资源获取即初始化(RAII)管用语法,并提供一个类,在它的析构函数中进行join。如下所示:

#include <iostream>
#include <thread>

// 新线程需要完成的操作
void ThreadPrint()
{
    std::cout << "A New Thread!" << std::endl;
}

// 异常处理
class Thread_Guard
{
private:
    std::thread& m_thread;
public:
    // 显式构造、传入引用
    explicit Thread_Guard(std::thread& p_thread):
        m_thread(p_thread) 
    {}

    // 析构、同步
    ~Thread_Guard()
    {
        if (m_thread.joinable())    m_thread.join();
    }

    // 禁止拷贝与赋值
    Thread_Guard(Thread_Guard const&) = delete;
    Thread_Guard& operator=(Thread_Guard const&) = delete;
};

// main()中调用client()
void client()
{
    // 新线程
    std::thread th(ThreadPrint);
    Thread_Guard thG(th);

    std::cout << "Client" << std::endl;
}

int main()
{
    client();
    return 0;
}

  在当前线程(main()所在的初始线程)运行到函数client()末尾时,局部对象thG会按照构造函数的逆序被析构,因此Thread_Guard对象thG首先被销毁,并且析构函数中的线程被结合。
  如果无需等待线程完成,可以通过分离(Detaching)来避免这种异常引起的安全问题。这打破了线程与std::thread对象的联系,并确保党std::thread对象被销毁时,std::terminate()不会被调用,即线程仍然在后台运行。

1.4 在后台运行线程

  在std::thread对象上对用detach()会把线程丢在后台运行,也没有直接的方法与之通信。也不可能等待该线程完成;如果一个线程成为分离的,获取一个引用它的std::thread对象也是不可能的,所以它也不可能再被结合。分离的线程在后台运行,其所有权和控制器交由C++的运行时库,以确保与线程相关的资源在线程退出后能被正确回收。

二、传递参数给线程函数

void func(int param);
std::thread th(func, 5);

  如上所示,我们可以通过以额外参数的方式将函数参数传给std::thread对象。值得注意的是,它默认以拷贝(Copied)方式传入。这意味着,我们在传入指针时需要注意类型转换;传入引用时需要使用std::ref。

2.1 避免隐式类型转换

  现在我们有一个C-style的string,我们期望调用它,但是我们不应该直接传入它的指针,而应该传入一个std::string对象:

char buffer[1024];
// do sth
std::thread th(func, std::string(buffer));
// instead of th(func, buffer);

2.2 传入引用时使用std::ref

  如果我们的函数明确接受一个引用参数,那我们在传入的时候最好使用std::ref。

2.2 thread的构造方式

  std::thread的构造方式和std::bind类似,具体可以参考这篇文章

三、转移线程所有权

  考虑下面两种情况:

  1. 编写一个函数,它创建一个在后台运行的线程,但是向调用函数传回新线程的所有权,而非等待其完成。
  2. 创建一个线程,并将所有权交给等待它完成的函数。

3.1 在thread对象之间转移

  与c++中的其他movable类型(如,std::ifstream, std::unique_ptr)一样,一个特定执行线程的权限可以在std::thread实例之间移动。下面一个例子展示了创建两个线程,以及在三个std::thread对象之间转移所有权:

void func_1();
void func_2();

std::thread th1(&func_1);
// 创建一个新线程,该线程拥有执行func_1的权限,同时该线程被th1持有。
std::thread th2 = std::move(th1);
// 移交th1持有线程的权限给th2,现在th2对象持有的线程负责执行func_1,th1失去原有线程的所有权。

/*
    th1: nullptr
    th2: &func_1
*/

th1 = std::thread(&func_2);
// 启动一个新的线程,并将它交给一个匿名thread对象,再将它交给th1,
// 这里不需要使用move语义,因为从临时对象中的move是自动且隐式的。
std::thread th3;
// 默认构造th3,它不予任何线程关联。
th3 = std::move(th2);
// 现在将th2的线程移交给th3,th3持有与func_1相关的线程,th2不与任何线程相关。

/*
    th1: &func_2
    th2: nullptr
    th3: &func_1
*/

th1 = std::move(th3);
// 将th3持有的线程交给th1,但是注意,这里th1已经持有了一个线程,
// 这将导致th1调用std::terminate()来终止程序,这样做是为了与std::thread的析构函数保持一致,
// 你必须在析构前等待线程完成或分离,同样:
// 在赋值时,不能通过向一个管理线程的thread对象赋一个新值来“舍弃”一个线程。

  下面来看上面例子的完整版:

#include <iostream>
#include <thread>

void func_1()
{
    std::cout << "1" << std::endl;
}

void func_2()
{
    std::cout << "2" << std::endl;
}

int main()
{
    std::cout << "begin" << std::endl;

    std::thread th1(&func_1);
    std::thread th2 = std::move(th1);

    th1 = std::thread(&func_2);
    std::thread th3;
    th3 = std::move(th2);

    th1 = std::move(th3);

    std::cout << "end" << std::endl;

    return 0;
}

  以下是输出结果:

begin
1
2
terminate called without an active exception

3.2 在函数之间转移

  函数间的线程转移与std::bind相同,这里不再赘述。但是我们可以通过这一点构造一个类似于thread_guard类,该类实际活得线程的所有权。

#include <iostream>
#include <thread>

class scoped_thread
{
private:
    std::thread m_th;

public:
    // 使用Move语义的显式构造
    explicit scoped_thread(std::thread p_th):
        m_th(std::move(p_th))
    {
        if (!m_th.joinable())   throw std::logic_error("No thread");
    }

    // 析构时join
    ~scoped_thread()
    {
        m_th.join();
    }

    // 禁止拷贝与赋值构造
    scoped_thread(scoped_thread const&) = delete;
    scoped_thread& operator=(scoped_thread const&) = delete;
};

void func(int num)
{
    std:: cout << num << std::endl;
}

int main()
{
    scoped_thread th(std::thread(&func, 5));

    return 0;
}

3.3 生成一批线程并等待完成

  考虑这样一个例子,我们给出10个线程,并打印一组数据:

#include <iostream>
#include <thread>
#include <cmath>
#include <vector>

/* ===== 积分计算函数 ===== */
void func(int id)
{
    std::cout << id << std::endl;
}

/* ===== main ===== */
int main()
{
    std::vector<std::thread> threads;
    const int threads_num = 10;

    // 创建线程
    for (int i = 0; i < threads_num; ++i)
    {
        threads.push_back(std::thread(&func, i));
    }

    double ans{ 0 };

    for (auto i = threads.begin(); i != threads.end(); ++i)
    {
        (*i).join();
    }   

    return 0;
}

  但是值得注意的是,这不是并行的任务,而是需要等待前一个线程完成的多线程顺序任务。

四、选择线程数量

  使用std::thread::hardware_concurrency()可以返回计算机的物理线程数量。

五、获取线程ID

  使用std::thread::id完成,对于std::thread对象可以调用成员函数get_id()实现。