前言

笔者在做 webserver 项目的时候，对每个连接的请求都是单线程运行，这样实现起来虽然简单，但是在连接数过多的时候处理速度会出现明显的下降，表现为每个连接的回应都有了一定的延迟。

解决方法是利用 cpu 的多核性能，并行处理每个连接的请求。并行处理可以使用多进程来实现，但是过多的请求会有比较大的上下文切换开销，对于这个 web 服务器，采用多线程明显是更好的方式。但是笔者还未学习过线程的知识，故在本文对学习的过程进行记录和总结。

学习的教材是C++ Concurrency In Action，本文所展示的代码全部在个人项目Learncpp的 Thread.cpp 文件中，欢迎参考。

正文

简单线程的创建和使用

基础

首先，要创建一个线程，我们需要包含 c++11 引入的头文件库<thread>,并以一个对象的管理方式创建 std::thread 对象来帮助我们管理这个线程。

其次，我们创建一个线程，是想要这个线程执行一定的工作。新进程在 fork 后和原进程在相同指令寄存器位置开始执行，而线程则是需要我们给这个线程指定一个函数来执行，可以理解为主线程从 main 函数进入，而子线程从我们指定的函数进入。这个指定函数的方式是给出函数指针。

#include <thread>
void hello(int id)
{
    cout << "hello thread" << id << '\n';
}
class Hello
{
    int value{};

public:
    void hello(const int i) const
    {
        cout << "hello from " << value + i << std::endl;
    }

    Hello() = default;

    explicit Hello(int i) : value(i) {};
};
void run()
{
    std::thread t1(hello , 1);//1
    t1.detach();//4
    int i{};
    std::thread t2([&i]
    {
        cout << "hello thread2" << i++ << std::endl;
    });//2
    t2.join();//4
    cout << i << '\n';
    Hello tem{10};
    std::thread t3(&Hello::hello , tem , 1);//3
    t3.join();
}

hello thread1
hello thread20
1
hello from 11

这里展示了 3 种函数传递类型。

$1$：hello()是我们定义的一个普通函数，在创建线程 t1 时我们直接写函数名称hello进去，编译器可以隐式的转换其为函数指针，当然我们也可以显式的写&hello。

提一嘴，这里更好的初始化是使用列表初始化{}来初始化线程，因为教材上用的是直接初始化，所以选择了保留这种初始化方式。（其实是写的时候忘了）

当函数的参数列表不为空，我们需要在构造时传入需要的函数参数，按顺序填入即可，和普通函数调用一样。

$2$：这里我们传入了一个匿名 lambda 函数。我们知道 lambda 函数实际会被编译器展开为一个结构体，这里&i构造引用成员i，并给出了一个调用运算符()的定义。

std::thread的构造函数模板参数第一个其实是一个可调用对象，所以重载了调用运算符的任何对象都可以传入std::thread构造函数，并通过参数类型包对参数进行打包和解包调用。这里t2线程执行完毕后可以看到主线程的i值发生了改变（由 0 变成了 1 ）。

$3$：在这里，我们创建了一个类，并想让线程t3执行类的成员函数。这里的情况与上方的情况有所不同，因为类的成员函数并不能隐式转换成函数指针，所以我们需要显式给出其地址&Hello::hello。

然后，因为成员函数需要一个类对象来调用，所以我们还需要传入一个类实例，这里有两种传入方式，std::thread的构造函数默认只会对每个参数进行复制，所以我们这里传入的其实是tem的一个副本，主线程中的tem并没有改变。要传入tem本身，我们需要传入其指针&tem。

这里对成员函数参数，可以理解为，普通的成员函数参数中隐含了一个this指针指向当前对象，所以需要传入一个对象实例。举一反三，由于静态成员函数没有这个指针，所以直接像普通函数一样调用即可。

$4$：对于所有线程，我们必须保证其在作用域内调用了join()或detach()函数，否则其在离开作用域析构的时候会抛出std::terminate异常。

join表示阻塞运行，主线程会一直等待子线程运行完毕再向下运行，以保证子线程已经完成了我们要求的工作；而detach分离运行，主线程和子线程直接分离，各干各的。这有时候会在进行 io 操作的时候发生一些意外，因为 io 操作非常耗时，两个线程同时进行 io ，会造成内容错位，比如有时上面代码的输出会是这样：

hello threadhello thread21
0
1

使用须知

这里给出几个使用线程的小知识。

void runthread(std::thread t)
{
    cout << "In runthread";
    t.join();
}
void run()
{
    int value = 10;
    std::thread t4([] (int& i) { i += 15; } , std::ref(value));    //1
    t4.join();
    cout << "t4 value: " << value << std::endl;
    std::thread t5 = std::move(std::thread(hello , 5));    //2
    if (t5.joinable()) {    //3
        runthread(std::move(t5));
    }
    if (t5.get_id() == t4.get_id()) {    //4
        cout << "t5 == t4" << '\n';
    }
}

$1$：当我们的函数参数需要一个引用时，我们将不能简单的传入&value，因为之前说过，std::thread对每个参数都进行的是复制，传入&value实际是传入了一个指针，所以这里我们需要std::ref()来帮助我们传入引用。

$2$：std::thread删除了复制构造和复制赋值函数，但是保留了其移动的能力，所以我们可以像移动std::unique_ptr一样移动std::thread，这里就移动了一个临时对象（虽然这个例子看起来挺多此一举），也可以用来实现函数传参。

$3$：std::thread提供了一个函数joinable()来指示其是否有被阻塞或分离运行，因为join或detach函数只能调用一次，所以为防止多次调用可以进行一个判断。

这里给出std::thread的析构函数。

~thread()
{
if (joinable())
std::terminate();
}

$4$：std::thread为每个线程分配了一个独一无二的 id 来标识，我们可以用get_id()来获取这个 id ，如果两个线程的 id 相同，那么可以说这两个线程要么是同一个线程，要么都没有线程。

线程的一个使用例子

得益于线程的并行执行可以极大的提高效率，其可以被用来提高相似运算的运行速度。下面是一个利用线程并行来计算 1 到 100 万的加法的例子。

//Iter迭代器类型，T返回值类型
template <typename Iter , typename T>
struct accumulate_block  //1
{
    void operator()(Iter first , Iter last , T& result)
    {
        result = std::accumulate(first , last , result);
    }
};

template <typename Iter , typename T>
T parallel_accumulate(Iter first , Iter last , T init)
{
    auto length = last - first;
    if (!length) {
        return init;
    }
    const int min_work_per_thread = 100;
    const int max_threads = (length + min_work_per_thread - 1) / min_work_per_thread;
    const int hardware_threads = static_cast<int>(std::thread::hardware_concurrency());
    const int num_threads = std::min(hardware_threads != 0 ? hardware_threads : 2 , max_threads);  //2
    const int block_size = length / num_threads;
    std::vector<T> result(num_threads);
    std::vector<std::thread> threads(num_threads - 1);//3
    Iter block_start = first;
    for (int i = 0 ; i < num_threads - 1 ; ++i) {
        Iter block_end = block_start;
        std::advance(block_end , block_size);
        threads[i] = std::thread(accumulate_block<Iter,T>() , block_start , block_end , std::ref(result[i]));
        block_start = block_end;
    }  //4
    accumulate_block<Iter,T>()(block_start , last , result[num_threads - 1]);
    std::for_each(threads.begin() , threads.end() , std::mem_fn(&std::thread::join));  //5
    return std::accumulate(result.begin() , result.end() , init);
}

void run()
{
    cout << "hardware threads: " << std::thread::hardware_concurrency() << std::endl;
    constexpr long long max_num = 1000000;
    std::vector<long long> nums(max_num);
    for (int i = 0 ; i < max_num ; ++i) {
        nums[i] = i;
    }
    auto start = std::chrono::high_resolution_clock::now();
    auto t = parallel_accumulate(nums.begin() , nums.end() , 0LL);
    auto end = std::chrono::high_resolution_clock::now();
    cout << t << " in parallel " << (end - start).count() << " ms" << std::endl;
    start = std::chrono::high_resolution_clock::now();
    t = std::accumulate(nums.begin() , nums.end() , 0LL);
    end = std::chrono::high_resolution_clock::now();
    cout << t << " in common " << (end - start).count() << " ms" << std::endl;
}

$1$：首先定义一个计算模块accumulate_block来对std::accumulate进行封装，因为std::accumulate的结果靠返回值获取，而线程的返回值需要更加深入的机制，所以我们把返回值绑定到第三个参数，靠输出参数方式获取结果。

$2$：其次，我们需要获知待计算的数据量，并设置每个线程最少的计算量，以选择合适的线程数执行计算操作。另外，我们通过std::thread::hardware_concurrency获取当前 cpu 可用核数，作为我们的线程上限。这里计算子线程的数据量时要记得主线程也可作为一个计算线程，用来计算尾块。

$3$：这里创建一个数组用来放置每个线程计算的结果，创建一个线程数组来对子线程进行管理。

$4$：创建子线程并分配任务，同步移动未计算迭代器。

$5$：对每个线程调用std::mem_fn(&std::thread::join)，使其运行完毕，std::mem_fn返回一个可调用的对象供每个迭代器调用。

最后对每个线程的结果相加便可得到最后的结果。运行并计时，有：

hardware threads: 24
499999500000 in parallel 1576716 ms
499999500000 in common 2933739 ms

可以看出并行在大数据的时候比单线程快了一倍。这个结论在小数据的时候不成立，因为线程的创建需要更大的开销。

数据的竞争和保护

因为多个线程共享大部分资源，所以会产生一些单线程没有的问题，其中数据竞争和条件竞争是最为常见的问题。这里先说明数据竞争。

constexpr int MAX = 10000;
std::vector<int> nums(MAX);
std::mutex nums_mutex;

void add(int index , int value)
{
    for (int i = 0 ; i < 100 ; ++i) {
        nums[index] += value;
    }
    for (int i = 0 ; i < 99 ; ++i) {
        nums[index] -= value;
    }
}

void add_mutex(int index , int value)
{
    nums_mutex.lock();
    add(index , value);
    nums_mutex.unlock();
}

void add_lock_guard(int index , int value)
{
    std::lock_guard<std::mutex> lock(nums_mutex);
    add(index , value);
}

using add_fun = void (*)(int index , int value);

void run_add(add_fun f , int index , int value)
{
    for (int i = 0 ; i < MAX ; ++i) {
        std::thread t(f , index , value);
        t.detach();
    }
    sleep(1);
    cout << nums[index] << '\n';
}


void run()
{
    run_add(add , 0 , 1);             //1
    run_add(add_mutex , 1 , 1);       //2
    run_add(add_lock_guard , 2 , 1);  //3
}

$1$：我们定义了一个全局的数组nums，在函数add中，我们对数组中一个数先加 100 次，再减 99 次value，最终结果应该是只加一次value，运行 10000 次，最终结果应该是 10000 ，但是实际运行结果是：

10209

这个数字每次运行的结果都会变化，其错误的原因是多个线程产生了数据竞争。想象这样一种情况，线程 A 读取数据为 100 ，在 cpu 中运算的时候线程 B 对数据进行读取，也是 100 ，最后 A 写入 99 ，而 B 也写入 99 ，两个线程本应相互协作写入 98 ，但是因为运算和 io 的延迟，第一次运算的数据被覆盖了，导致最后没有减到相应的数值。

解决办法是设置一个变量，让一个线程使用数据时设置这个变量为否，然后其他线程只能等待这个变量为真才能继续执行他们的操作。用bool类型的变量在这里行不通，因为这种基本类型没有特殊保护，会像上面的数据竞争一样产生冲突。这里，我们使用一种叫互斥量的东西，类型为std::mutex，并引入一种由库和操作系统共同实现的锁机制。

$2$：在这个带锁版本的函数中，我们先对这个互斥量上锁，然后执行我们的操作，最后对互斥量解锁。在互斥量上锁期间，只有对该互斥量上锁的线程可以活动，其他线程会阻塞到上锁的函数中，一直等到解锁并轮到自己上锁的时刻。底层实现保证了这一点，所以我们解决了数据竞争，不会有多个线程同时对同一数据进行读写。不过锁机制会增加开销，使程序的运行变慢，所以只在必要的时候使用互斥量上锁。

这里必须保证互斥量的作用域大于要保护的数据的作用域，否则相同或不同的函数可能会持有不同的互斥量，从而让保护机制失效。

$3$：可以看出，例 2 的互斥量上锁和解锁需要我们自己来管理，这不是一个好现象，因为我们很可能会忘记解锁，就像我们new之后忘记delete一样。根据 RAII 原则，我们可以用一个类来帮我们自动管理互斥量，幸运的是，std提供了现成的这样的类：std::lock_guard。std::lock_guard是一个模板类，模板参数接收一个互斥量类型，并且创建实例时需要给出一个需要管理的互斥量，他会自动帮助我们上锁，并在退出作用域时自动解锁。

测试代码输出：

10209
10000
10000

条件竞争

有时候我们发现，即使我们对数据进行了保护，多线程之间也可能会发生条件竞争，这是由于操作不是原子性导致的。 学习了数据竞争后，我们有了这样一个数据安全的栈：

class safe_stack  
{  
    std::vector<int> nums{};  
    std::mutex nums_mutex;  
    int id{};  
  
public:  
    [[nodiscard]] int getid()  
    {  
        std::lock_guard<std::mutex> lock(nums_mutex);  
        return id;  
    }
  
    safe_stack(int i) : id{i} {};  
  
    void push(int value)  //1
    {  
        std::lock_guard<std::mutex> lock(nums_mutex);  
        nums.push_back(value);  
        cout << "push " << value << '\n';  
    }  
  
    bool empty()  //2
    {  
        std::lock_guard<std::mutex> lock(nums_mutex);  
        return nums.empty();  
    }  
  
    void pop(int& value)  //3
    {  
        std::lock_guard<std::mutex> lock(nums_mutex);  
        if (nums.empty()) {  
            cout << "empty!" << '\n';  
            return;  
        }  
        value = nums[nums.size() - 1];  
        nums.pop_back();  
        cout << "pop " << value << '\n';  
    }  
  
    std::shared_ptr<int> pop()  //4
    {  
        std::lock_guard<std::mutex> lock(nums_mutex);  
        auto t = std::make_shared<int>(nums[nums.size() - 1]);  
        nums.pop_back();  
        return t;  
    }  
};
void run()  
{  
    safe_stack stack{1};  
    auto pop = static_cast<void(safe_stack::*)(int&)>(&safe_stack::pop);  
    for (int i = 0 ; i < 100 ; ++i) {  
        if (stack.empty()) {  
            std::thread t(&safe_stack::push , &stack , i);  
            t.detach();  
        }  
        else {  
            int value{};  
            std::thread t(pop , &stack , std::ref(value));  
            t.detach();  
        }  
    }
}

一切看起来很完美。我们的每个接口都使用了作为成员变量的互斥量来进行保护，然后我们检查这个栈，如果是空的就推入数据，否则弹出数据。那么运行他吧：

...
pop 0
empty!
push 7
...

程序出错了！按照我们的分析，这个程序不应该打印empty，到底是哪里出了问题？ 你可能发现了，我们我们在检查栈为空后和执行操作前这一段时间是没有互斥量保护的

总结