多线程并发编程

一、多线程带来的问题

以线程对一个全局变量进行自减为例引入多线程带来的问题。该操作分为三步：

1. 将变量从内存移动到CPU相关寄存器
2. 对它进行--操作
3. 将--后的值写回内存

但是当系统进行内核态->用户态转换时，可能会进行线程的切换。倘若一个进程的--操作在执行到第2步时被切换，让别的线程再次操作这个全局变量，此时该变量的值再次改变。那么下一次线程切换回来时，寄存器中暂存的值被写回内存，从而导致数据不一致的问题。

相关概念

临界资源：多线程共享的全局变量等资源

临界区：每个线程内部，访问临界资源的代码，就叫做临界区

原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成

为了保证多线程访问临界资源不会引发数据不一致等问题，线程库引入了互斥与同步的概念。

二、互斥

1、互斥与互斥量

互斥：在任何时刻，保证有且只有一个线程进入临界区，访问临界资源。

为了完成互斥操作，库引入了互斥量的定义。

互斥量又被形象地称为==“锁”==，在进入临界区前加锁是每一个线程必须遵循的规定！

• 当一个线程竞争到互斥量时，该互斥量会被锁定，其余竞争互斥量的线程此时会因为竞争失败而陷入阻塞状态，暂时被挂起。
• 当竞争到互斥量的线程将互斥量解锁后，其余线程才能申请到。
• 倘若线程竞争到互斥量后因线程调度被切换走，那么该互斥量依然保持加锁状态，其余线程无法申请到互斥量，直到互斥量解锁。

若某线程在申请到互斥量后再次申请该互斥量，那么由于前一次申请将其加锁，第二次申请就会导致该线程被挂起，之后所有线程都无法申请到该互斥量，因此整个进程都会处于卡死状态(死锁)。

2、申请互斥量

I. 静态方法申请互斥量：

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER

通过一个宏来初始化互斥量。

注：通过宏初始化的互斥量不是同一个，因此在使用时，不会相互影响。

II. 动态方法申请互斥量：

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr)

• mutex：要初始化的互斥量
• attr：互斥量的设置参数，设置为NULL即可
• 返回值：成功则返回0，失败则返回错误码。

注：互斥量为pthread_mutex_t类型的变量，需要用户定义，通过输出型参数的方式传给pthread_mutex_init()函数。

3、利用互斥量加锁与解锁

int pthread_mutex_lock(pthread_mutex_t *mutex) // 加锁

int pthread_mutex_unlock(pthread_mutex_t *mutex) // 解锁

成功则返回0，失败则返回错误码。

4、销毁互斥量

int pthread_mutex_destroy(pthread_mutex_t *mutex)

成功则返回0，失败则返回错误码。

注：动态方法申请的互斥量必须手动销毁。

5、互斥量综合应用——模拟抢票

// 模拟抢票
int tickets = 100;
pthread_mutex_t lock;

void* routine(void* arg)
{
    while (1)
    {
        // 进入临界区前先加锁
        pthread_mutex_lock(&lock);
        if (tickets > 0)
        {
            pthread_mutex_lock(&lock);
            usleep(20000); // 模拟购票成功后的系统响应时间
            cout << pthread_self() << " booking success! tickets left: " << tickets << endl;
            tickets--;
            pthread_mutex_unlock(&lock); // 锁不用了之后一定要解锁，因此每一个分支都要unlock
        }
        else 
        {
            pthread_mutex_unlock(&lock);
            break;
        }
    }
}

const int NUM = 5; // 线程数

int main()
{
    pthread_mutex_init(&lock, nullptr);
    pthread_t tids[NUM];
    for (int i = 0; i < NUM; ++i)
    {
        pthread_create(tids + i, nullptr, routine, nullptr);
        cout << tids[i] << "thread created successfully" << endl;
    }

    
    for (int i = 0; i < NUM; ++i)
    {
        pthread_join(tids[i], nullptr);
        cout << tids[i] << "thread quited successfully! tickets left = " << tickets << endl;
    }
    
    pthread_mutex_destroy(&lock);

    return 0;
}
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

6、互斥量的原子性

由于有多个线程需要竞争互斥量，因此互斥量本身也是一个临界资源，但是它具有原子性，其加解锁模型可被简化如下：

1. 互斥量mutex存储在内存中，它的初始值为1。

2. 当线程通过lock想申请互斥量时，首先要将自己标识是否持有锁的寄存器初始化为0，然后让该寄存器的内容与mutex进行交换。若交换后寄存器为1，则说明该线程申请到了互斥量，同时mutex由于交换变为0，说明线程将其锁住了。

3. 如果此时该线程被切换，那么由于mutex已经变成0，此时其他线程就永远得不到1了，而原先的那个1还存储在被切换的线程的上下文数据中。

4. 对于解锁操作，将mutex重新初始化为1，然后唤醒其它因申请互斥量被阻塞的线程即可。 对于汇编语言，将寄存器内容与内存地址上的内容交换是一条XCH指令就可以完成的，因此lock是具有原子性的。

7、线程饥饿问题

当多个竞争的线程被设置优先级之后，优先级越高，线程被给予的CPU时间越多。但是在某些极端情况下，比如存在多个高优先级线程，那么低优先级的线程可能永远无法被授予充足的CPU时间，从而导致这些线程处于==“饥饿”==状态。

三、同步

1、同步与条件变量

同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源。这种顺序往往基于一种依赖关系，比如：A线程负责生产数据，B线程负责处理A生产的数据，那么在当前B没有数据可处理的情况下必须满足A在B之前执行。

为了完成同步操作，库引入了条件变量的定义。

条件变量与互斥量搭配使用，使得线程以顺序等待而非竞争的方式访问临界资源。

2、条件变量的初始化

int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr)

• cond：要初始化的条件变量
• attr：条件变量的设置参数，设置成NULL即可
• 成功则返回0，失败则返回错误码。

3、线程等待满足条件变量

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex)

• cond：等待条件变量cond被满足
• mutex：互斥量
• 成功则返回0，失败则返回错误码。

该函数的作用是：

1. 将mutex释放。若不将mutex释放，而选择带着mutex一起被挂起，那么其它需要mutex进入临界区的线程都将陷入阻塞。

2. 使本线程陷入阻塞，等待被唤醒。

3. 被唤醒后，重新获取mutex。因为该线程依然处于临界区，所以被唤醒后需要重新获取锁以保证临界资源的安全。

注：Linux为每个条件变量维护了一个等待队列，当前被阻塞的线程被添加到该队列当中，等待被唤醒。

4、唤醒进程

解除一个线程由于等待条件变量满足而被阻塞的状态。若等待队列有多个线程，则具体解除哪一个由操作系统的调度算法决定。

I. 唤醒一个等待的线程

int pthread_cond_signal(pthread_cond_t *cond)

II. 唤醒所有等待的线程

int pthread_cond_broadcast(pthread_cond_t *cond)

5、条件变量的销毁

int pthread_cond_destroy(pthread_cond_t *cond)

6、条件变量的经典应用——生产者消费者模型

在生产者-消费者模型中，生产者Producer负责生产数据，而消费者Consumer负责获取并处理生产者生产的数据。多个生产者线程会在同一时间运行，生产数据；同时，也会有多个消费者线程同时运行，获取并处理数据。

产出的数据被存放在内存的某一个区域(即**“仓库”**，具体实现就是数组、队列…)。生产者只需要关心是否需要继续生产数据并将其存入仓库，而消费者也只需要关心仓库中是否有数据并将数据取走。

因此，该“仓库”的作用是：将生产者与消费者进行解耦，使其只需要关心自己的行为，相互之间没有影响，那么维护的成本也就更低了。

I. 生产者与生产者、消费者与消费者的关系

他们都是互斥关系。

生产者之间竞争生产数据的资格，消费者之间竞争内存中的数据资源。本质上，都是竞争进入临界区的锁。

II. 生产者与消费者的关系

他们是互斥与同步的关系。

互斥体现在：生产数据和使用数据这两种行为不能同时发生，即临界区一次只能由一个生产者线程或消费者线程进入。

同步体现在：当==“仓库”为空==，消费者需要等待生产者产出数据后才能消费。当==“仓库”为满==，生产者需要等待消费者取走数据后才能生产。即：在特定情况，它们需要被阻塞以等待对方的唤醒。

III. 代码实现

// 模板参数T表示生产者生产的数据类型
template
class ProducerConsumer
{
private:
    std::queue bq;           // 仓库(这里用队列实现)
    int capacity;               // 仓库的容量(即队列的最大长度)
    pthread_mutex_t lock;       // 读写仓库的互斥量
    pthread_cond_t can_produce; // 生产者可以生产的条件变量（仓库为空）
    pthread_cond_t can_consume; // 消费者可以消费的条件变量（仓库不为空）
private:
    bool isEmpty()
    {
        return bq.empty();
    }
    
    bool isFull()
    {
        return bq.size() == capacity;
    }
public:
    ProducerConsumer(int cap) : capacity(cap) 
    {
        // 初始化互斥量与条件变量
        pthread_mutex_init(&lock, nullptr);
        pthread_cond_init(&can_produce, nullptr);
        pthread_cond_init(&can_consume, nullptr);
    }

    void produce(const T& data)
    {
        // 进入临界区前先上锁
        pthread_mutex_lock(&lock);
        while (isFull()) // while循环，避免伪唤醒，即：本线程醒来后被切换出去，其它线程取走数据，本线程回来的时候又没数据了
        {
            pthread_cond_wait(&can_produce, &lock); // 阻塞，等待被唤醒
        }
        
        // 走到这里说明有空间可以存放生产的数据
        bq.push(data);
        pthread_mutex_unlock(&lock); // 出临界区，解锁
        pthread_cond_signal(&can_consume); // 通知消费者可以消费

    }

    void consume(T* out)
    {
        // 进入临界区前先上锁
        pthread_mutex_lock(&lock);
        while (isEmpty())
        {
            pthread_cond_wait(&can_consume, &lock); // 等待消费
        }

        // 走到这里说明有数据供消费
        T data = bq.front();
        bq.pop();
        *out = data;

        pthread_mutex_unlock(&lock); // 出临界区，解锁
        pthread_cond_signal(&can_produce); // 通知生产者继续生产
    }

    ~ProducerConsumer()
    {
        // 销毁互斥量与条件变量
        pthread_mutex_destroy(&lock);
        pthread_cond_destroy(&can_produce);
        pthread_cond_destroy(&can_consume);
    }
};
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

四、线程池

线程池是一种线程的使用模式，其内部维护着多个线程，等待分配可并发执行的任务。

1、线程池的优势

1. 避免了在处理短时间任务时创建与销毁线程的代价。

2. 可以调节线程池中的线程数量，防止因为消耗过多内存而导致系统崩溃。

注：可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

2、线程池的应用场景

1. 需要大量的线程来完成任务，且完成任务的时间比较短的程序。 比如WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大。

2. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求在没有线程池的情况下，将使服务器短时间内产生大量线程，使内存到达极限，出现错误。

五、可重入与线程安全

线程安全：多个线程并发执行同一段代码时，产生的结果与预期相同。

重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数。

1、线程不安全的函数

• 不保护共享变量(临界资源)的函数。

• 函数状态随着被调用，状态发生变化，比如其内部有static变量。

• 返回指向静态变量的指针的函数：因为这个静态变量可以被所有调用该函数的线程得到，从而在函数外也变成临界资源。

• 调用了线程不安全函数的函数。

2、不可重入的函数

• 调用了malloc/free函数：因为malloc函数是用全局链表来管理堆上的内存的。
• 调用了标准I/O库函数：因为标准I/O库的很多实现都使用了全局的数据结构。
• 使用了静态的数据结构

如果某个函数的参数全部是值传递，且函数内只是用栈上的局部变量，那么该函数就可以被断言为可重入函数。

3、可重入与线程安全的区别

1. 线程安全不一定是可重入的，但可重入函数则一定是线程安全的。
2. 线程安全依赖于互斥量和条件变量，而可重入函数不需要任何同步与互斥的操作，因此更加高效。
3. 如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数的锁还未释放则会产生死锁，因此是不可重入的。

4、STL与线程安全

STL不是线程安全的，因为STL的设计初衷是将性能挖掘到极致，一旦涉及到加锁保证线程安全，会对性能造成巨大的影响。此外，对于不同的容器，加锁方式的不同，性能可能也不同。

结论：如果需要在多线程环境下使用STL，需要用户自行保证线程安全！

六、死锁

死锁：是指两个及以上的线程在执行过程中，因争夺资源而造成的一种互相等待的现象。

1、死锁四个必要条件

1. 互斥：涉及的资源是非共享的，即资源被一个线程占用时，其他线程无法使用。

2. 不可剥夺:线程已获得的资源，在末使用完之前，不得被其它线程强行剥夺。

3. 请求与保持：线程请求新资源的同时对已获得的资源保持不放。

4. 循环等待:若干线程之间形成一种头尾相接的循环等待资源的关系。

2、如何避免死锁

• 破坏死锁的四个必要条件中的一个或多个。

• 线程使用互斥量进行加锁的顺序要一致 。

• 避免使用锁但不释放。

七、特殊类型的锁

1、自旋锁

对于普通的互斥锁，当互斥量被锁住时，其余申请锁的线程会被挂起，然而唤醒并切换一个被挂起的线程是有较大开销的。如果让没申请到锁的线程通过循环不断地尝试获取锁，使得该线程一直处于运行状态而非被挂起，那么就能节省线程切换的开销。这就是自旋锁的概念。

注意：如果临界区较大，即其它线程拿到锁进入临界区后需要运行较长时间才会释放锁，那么循环获取锁相较于线程切换反而开销更大。因此，自旋锁只适用于临界区较小的情况。

2、读写锁(读者写者问题)

有些情况下，修改数据的次数很少，大多数都是读数据，而读往往伴随着耗时的查找，如果对读数据的代码段进行加锁，会使整个程序的效率大大降低。因此，读写锁应运而生，它使得一次只能有一个写线程，但是可以有多个读线程并发地读数据。

首先明确：

• 读者与写者之间存在互斥关系，因为写会影响读的内容；
• 写者与写者之间存在互斥关系，因为它们会互相影响
• 读者与读者之间不存在任何关系，因为它们都不会修改数据

读写锁就是一把锁，但是它有两种状态。

1. 当锁被读线程占用时，则所有读线程都可以共享这一把锁，即它们都可以进行读数据的操作，无需等待解锁而挂起。若写线程此时申请锁，则必须等当前所有读线程释放锁以后，才可以进行写操作。因此，可以理解成：读写锁维护了一个读线程计数器，记录当前有多少读线程正在使用这把锁，当计数器归零时，写线程才有权使用读写锁。

2. 当锁被写线程占用时，则其余写线程和读线程都将被阻塞。

读写锁接口

初始化

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr)

销毁

int pthread_rwlock_destroy(pthread_rwlock_t *rwlock)

加锁和解锁

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock)

int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock)

int pthread_rwlock_unlock(pthread_rwlock_t *rwlock)

3、乐观锁与悲观锁

I. 悲观锁

悲观锁在操作数据时比较"悲观"，每次去拿数据的时候都认为别的线程也会同时修改数据，所以每次在拿数据的时候都会上锁，这样别的线程想拿到这个数据就必须阻塞等待。

II. 乐观锁

乐观锁是对于数据冲突保持一种"乐观态度"，操作数据时不会对操作的数据进行加锁，只有到数据提交的时候才通过一种机制来验证数据是否存在冲突(一般实现方式是通过加版本号然后进行版本号的对比方式实现)。