【Linux】解锁并发：多线程同步技术详解与应用实践

前言：

在现代软件开发中，多线程编程已成为提高程序性能和资源利用率的关键技术。然而，多线程环境下的线程同步问题一直是开发者需要面对的挑战。本文旨在深入探讨多线程同步的基本概念、机制和实际应用，帮助读者理解并掌握如何在多线程环境中实现有效的线程间协作。

本文首先介绍了同步的基本概念，解释了同步在多线程编程中的重要性。随后，详细讨论了条件变量这一实现线程间同步的重要工具，并通过一个生动的果农与猴子的比喻，形象地说明了条件变量的工作原理和应用场景。接着，文章通过实际的测试代码示例，展示了如何使用条件变量和互斥锁来实现线程间的同步。

文章进一步探讨了生产消费模型、线程池、可重入与线程安全、死锁等多线程编程中的核心问题，并提供了相应的解决方案和最佳实践。特别地，对于线程安全的单例模式、自旋与阻塞挂起、以及读者写者问题等高级主题，文章不仅提供了深入的理论分析，还给出了具体的代码实现和应用示例。

1. 同步概念

同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步。

2. 条件变量：实现线程间同步的！

2.1. 条件变量是什么？

• 当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
• 例如一个线程访问队列时，发现队列为空，它只能等待，只到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量。

举个例子：

场景设定：

• 盘子：用来存放苹果的容器，初始时盘子里没有苹果。
• 果农：负责往盘子里放苹果的人。
• 猴子：等待盘子里有苹果后，才会从盘子里拿苹果吃。
• 铃铛：一个信号装置，果农放完苹果后会摇铃铛，以此通知猴子

行为流程：

1. 猴子等待： 猴子检查盘子里的苹果数量。 如果盘子里没有苹果，猴子就等待。
2. 果农放苹果： 果农检查盘子里的苹果数量。 如果盘子里的苹果少于设定的数量（比如5个），果农就往盘子里放一些苹果。 放完苹果后，果农摇铃，发出声音信号。
3. 猴子被唤醒： 猴子听到铃铛声后，知道盘子里已经有苹果了。 猴子再次检查盘子里的苹果数量，确认后开始吃苹果。
4. 猴子吃苹果： 猴子从盘子里拿一个苹果吃。 吃完后，猴子继续等待，重复之前的检查和等待过程。

条件变量的应用：
在编程实现中，铃铛可以类比为条件变量的通知机制。当果农（一个线程）放完苹果并摇铃铛后，条件变量会被触发，这相当于通知所有等待的猴子（其他线程）可以开始执行它们的任务（吃苹果）。

使用条件变量的好处是，猴子线程不需要不断地检查盘子里是否有苹果，这会浪费CPU资源。相反，猴子线程可以在条件不满足时挂起，直到被果农线程通过摇铃铛（条件变量通知）唤醒。

在实际的多线程编程中，我们会使用互斥锁来保护对盘子里苹果数量的访问，并使用条件变量来同步果农和猴子的行为。当果农放苹果时，它会锁定互斥锁，更新苹果数量，然后摇铃铛（使用条件变量通知等待的猴子）。猴子在等待时会锁定互斥锁，检查条件，如果不满足就挂起等待，直到听到铃铛声（被通知）后再次检查条件并执行任务。

条件变量 = 铃铛 + 队列

2.2. 认识条件变量接口

pthread_cond_t:

pthread_cond_t cond = PTHREAD_COND_INITIALIZER; // 全局，静态
1

• 初始化：

int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrictattr);
1

参数：
cond：要初始化的条件变量
attr：NULL

• 销毁：

int pthread_cond_destroy(pthread_cond_t *cond)
1

• 等待条件满足

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

参数：
cond：要在这个条件变量上等待
mutex：互斥量，后面详细解释

• 唤醒等待

int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒所在cond等待的线程
int pthread_cond_signal(pthread_cond_t *cond); // 唤醒一个线程
1
2

3. 写一个测试代码——验证线程的同步机制

#include <iostream>
#include <string>
#include <vector>
#include <pthread.h>
#include <unistd.h>

pthread_cond_t gcond = PTHREAD_COND_INITIALIZER; // 条件变量
pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER; // 互斥锁

void* SlaverCore(void* args)
{
    std::string name = static_cast<const char*>(args);
    while (true)
    {
        // 1. 加锁
        pthread_mutex_lock(&gmutex);

        // 2. 一般条件变量是在加锁和解锁之间使用
        pthread_cond_wait(&gcond, &gmutex); // gmutex:这个是，用来被释放的！
        std::cout << "当前被叫醒的线程是：" << name << std::endl;

        // 3. 解锁
        pthread_mutex_unlock(&gmutex);
    }
}

void* MasterCore(void* args)
{
    sleep(3);
    std::cout << "master 开始工作了" << std::endl;
    std::string name = static_cast<const char*>(args);
    while (true)
    {
        pthread_cond_signal(&gcond); // 唤醒其中一个队列首部的线程
        // pthread_cond_broadcast(&gcond); // 唤醒队列中所有的 线程
        sleep(1);
        std::cout << "master 唤醒一个线程" << std::endl;
    }
}

void StartMaster(std::vector<pthread_t> *tidsptr)
{
    pthread_t tid;
    int n = pthread_create(&tid, nullptr, MasterCore, (void *)"Master Thread");
    if (n == 0)
    {
        std::cout << "create master success" << std::endl;
    }
    tidsptr->emplace_back(tid);
}

void StartSlaver(std::vector<pthread_t> *tidsptr, int threadnum = 3)
{
    for (int i = 0; i < threadnum; ++i)
    {
        char *name = new char[64];
        snprintf(name, 64, "slaver-%d", i + 1); // thread-1
        pthread_t tid;
        int n = pthread_create(&tid, nullptr, SlaverCore, name);
        if (n == 0)
        {
            std::cout << "create success" << name << std::endl;
            tidsptr->emplace_back(tid);
        }
    }
}

void WaitThread(std::vector<pthread_t>& tids)
{
    for (auto &tid : tids)
    {
        pthread_join(tid, nullptr);
    }
}

int main()
{
    std::vector<pthread_t> tids;

    StartMaster(&tids);
    StartSlaver(&tids, 5);
    WaitThread(tids);

    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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

4. 生产消费模型

讨论：并发数据的传递问题。

• 举一个超市的例子：

我们可以将超市的仓库看作是缓冲区，超市的供应商（生产者）负责提供货物（生产数据），而顾客（消费者）则负责购买这些货物（消费数据）。以下是这个模型在超市中的具体应用：

• 缓冲区（仓库）：超市的仓库是一个有限的空间，用来存储供应商送来的货物。

• 生产者（供应商）：供应商定期向超市仓库提供货物，比如新鲜的水果、蔬菜、肉类等。

• 消费者（顾客）：顾客到超市来购买他们需要的商品。

• 同步机制：为了防止生产者在仓库已满时继续生产货物，或者消费者在仓库为空时尝试购买商品，需要有一套同步机制来控制生产者和消费者的行为。在超市中，这可以通过库存管理系统来实现，确保货物的供应和销售是协调的。

• 互斥锁：当一个供应商正在向仓库添加货物时，其他供应商需要等待，直到当前供应商完成添加。同样，当一个顾客正在购买商品时，其他顾客需要等待，直到该顾客完成购买。

• 条件变量：当仓库为空时，消费者（顾客）可以等待，直到生产者（供应商）提供新的货物。相反，当仓库已满时，生产者（供应商）可以等待，直到消费者（顾客）购买一些商品，为新货物腾出空间。

• 死锁避免：超市需要确保不会发生死锁，即供应商和顾客都在等待对方采取行动，导致双方都无法前进。这可以通过合理的库存管理和顾客流控制来避免。

超市：共享资源——临界资源
厂商s, 用户s: 多个线程

超市是什么？临时保存数据的"内存空间"——某种数据结构对象。是数据"交易"的场所。
商品是什么？数据

并发问题：
生产者 vs 生产者 —— 互斥
消费者 vs 消费者 —— 互斥
生产者 vs 消费者 —— 互斥 && 同步

“321”: 3种关系，2种角色，1个交易场所

为什么？生产消费模型，可以提供比较好的并发度
使用生产和消费数据，进行解耦
支持忙闲不均

5. 生产消费模型 + 条件变量

BlockingQueue
在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)

基于环形队列的生产消费模型 :

• 环形队列采用数组模拟，用模运算来模拟环状特性:
  
  当队列为空的时候：p c index 是同一个位置
  当队列为满的时候：p c index 是同一个位置

p c index 是同一个位置:
a. 队列满 ：访问临界资源，让消费者跑。
b.队列空：访问临界资源，生产者先跑。

生产者不能把消费者套一个圈，消费者不能超过生产者！

p c index 不是同一个位置：一定不为空 && 一定不为满

生产和消费动作，可以真正的并发吗？ 可以的

怎么做到呢？ 信号量

1. 伪代码
   a. 对于生产者，最关心的是空间
   b. 对于消费者，最关心的是数据

sem_t room(10); // 空间信号量
sem_t data(0);	// 资源信号量
1
2

生产者：

P(room); // 申请空间资源(空间信号量--)
// 信号量申请成功
ringbuffer[p_index] = data;
p_index++;
p_index %= 10;
V(data); // 资源信号量++
1
2
3
4
5
6

消费者：

P(data);  // 申请数据资源
// 信号量申请成功——数据资源一定有
out = ringbuffer[c_index];
c_index++; 
c_index %= 10;
V(room);
1
2
3
4
5
6

信号量接口：

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_destroy(sem_t *sem);

int sem_wait(sem_t *sem); // P操作
int sem_post(sem_t *sem); // V操作
1
2
3
4
5
6

如果是多生产者，多消费者呢？
解决方法：加锁。
加几把锁？ 2把
如何加锁?

#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
#include <pthread.h>

template<typename T>
class RingQueue
{
private:
    void P(sem_t &sem)
    {
        sem_wait(&sem);
    }
    void V(sem_t &sem)
    {
        sem_post(&sem);
    }
    void Lock(pthread_mutex_t &mutex)
    {
        pthread_mutex_lock(&mutex);
    }
    void Unlock(pthread_mutex_t &mutex)
    {
        pthread_mutex_unlock(&mutex);
    }
public:
    RingQueue(int cap): _ring_queue(cap), _cap(cap),  _productor_step(0), _consumer_step(0)
    {
        sem_init(&_room_sem, 0, _cap);
        sem_init(&_data_sem, 0, 0);

        pthread_mutex_init(&_productor_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }
    void Enqueue(const T &in)
    {
        // 生产行为
        P(_room_sem);
        Lock(_productor_mutex);
        // 一定有空间！！！
        _ring_queue[_productor_step++] = in; // 生产
        _productor_step %= _cap;
        Unlock(_productor_mutex);
        V(_data_sem);
    }
    void Pop(T *out)
    {
        // 消费行为
        P(_data_sem);
        Lock(_consumer_mutex);
        *out = _ring_queue[_consumer_step++];
        _consumer_step %= _cap;
        Unlock(_consumer_mutex);
        V(_room_sem);
    }
    ~RingQueue()
    {
        sem_destroy(&_room_sem);
        sem_destroy(&_data_sem);

        pthread_mutex_destroy(&_productor_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }
private:
    // 1. 环形队列
    std::vector<T> _ring_queue;
    int _cap; // 环形队列的容量上限

    // 2. 生产和消费的下标
    int _productor_step;
    int _consumer_step;

    // 3. 定义信号量
    sem_t _room_sem; // 生产者关心
    sem_t _data_sem; // 消费者关心

    // 4. 定义锁，维护多生产多消费之间的互斥关系
    pthread_mutex_t _productor_mutex;
    pthread_mutex_t _consumer_mutex;
};  
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
72
73
74
75
76
77
78
79
80
81

6. 线程池

一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

• 线程池的应用场景：

1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.

• 线程池的种类：
• 线程池示例：

1. 创建固定数量线程池，循环从任务队列中获取任务对象，
2. 获取到任务对象后，执行任务对象中的任务接口

7. 可重入 VS 线程安全

7.1. 概念

• 线程安全(线程执行中的互相关系)：多个线程并发同一段代码时，不会出现不同的结果。常见对全局或静态操作，并且没有锁保护的情况下，会出现该问题。
• 重入（函数的特点）：同一个函数被不同的执行流调用，当前一个流程还没有执行完成，就会有其他执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或任何问题，则该函数被称为可重入函数，否则，是不可重入函数。

7.2. 常见线程不安全的情况

• 不保护共享变量的函数
• 函数状态随着被调用，状态发生变化的函数
• 返回指向静态变量指针的函数
• 调用线程不安全的函数

7.3. 常见的线程安全的情况

• 每个线程对全局变量或者静态变量只有读权限，而没有写权限，一般来说这些线程是安全的
• 类或者接口对于线程来说都是原子操作
• 多个线程之间的切换不会导致接口的执行结果存在二义性

7.4. 常见不可重入的情况

• 调用了malloc/free函数，因为malloc函数是全局链表来管理堆的
• 调用了标准I/O库函数，标准I/O库的很多实现都是不可重入的方式使用全局数据结构
• 可重入函数体内使用了静态的数据结构

7.5. 常见可重入的情况

• 不使用全局变量或静态变量
• 不使用用malloc或者new开辟出来的空间
• 不调用不可重入函数
• 不返回静态或全局数据，所有数据都有函数的调用者提供
• 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据

7.6. 可重入与线程安全联系

• 函数是可重入的，那就是线程安全的
• 函数是不可重入的，那就不能由多个线程使用，可能引发线程安全问题
• 如果一个函数中有全部变量，那么这个函数既不是线程安全也不是可重入的。

7.7. 可重入与线程安全的区别

• 可重入是线程安全的一种。
• 线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
• 如果将临界资源的访问上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生死锁，因此是不可重入的。

8. 死锁

死锁是多线程把锁不合理的使用， 导致代码不会继续向后推进的情况。
有时候访问一块临界资源的时候需要同时持有两把锁才能访问，但是有两把锁互相持有一把锁，它们两互相申请对方的锁，就造成死锁了！
死锁的四个必要条件：

• 互斥条件：一个资源每次只能被一个执行流使用。
• 请求与保持条件：一个执行流因请求资源阻塞时，对已获得的资源保持不放。
• 不剥夺条件：一个执行流已经获得资源，在未使用完之前，不能前行剥夺。
• 循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系。

解决、避免死锁条件必定是破坏4个条件之一。建议大家如果申请多把锁，每个线程申请锁的顺序一致。

9. STL，智能指针和线程安全

9.1. STL中的容器是否是线程安全的？

不是，
原因是，STL的设计初衷是将性能挖掘打极致，而一旦涉及到加锁保证线程安全，会对性能造成巨大的影响。且对不同的容器，加锁的方式不同性能也可能不同，性能可能也不同（例如hash表的锁表和锁桶）。
因此 STL 默认不是线程安全，如果需要在多线程环境下使用，往往需要调用者自行保证线程安全。

9.2. 智能指针是否线程安全的？

对于 unique_ptr, 由于只是当前代码块范围内生效，因此不涉及线程安全问题。
对于shared_ptr, 多个对象需要共用一个引用计数变量，所以会存在线程安全问题，但是标准库实现的时候考虑了这个问题，基于原子操作（CAS）的方式保证 shared_ptr 能够高效，原子的操作引用计数。

10. 线程安全的单例模式

10.1. 什么是单例模式

单例模式是一种“经典的，常用的，常考的” 设计模式。

10.2. 什么是设计模式

IT行业这么火, 涌入的人很多. 俗话说林子大了啥鸟都有. 大佬和菜鸡们两极分化的越来越严重. 为了让菜鸡们不太拖大佬的后腿, 于是大佬们针对一些经典的常见的场景, 给定了一些对应的解决方案, 这个就是设计模式。

10.2. 单例模式的特点

某些类, 只应该具有一个对象(实例), 就称之为单例。
例如一个男人只能有一个媳妇。
在很多服务器开发场景中,经常需要让服务器加载很多的数据 (上百G) 到内存中。此时往往要用一个单例的类来管理这些数据。

10.3. 饿汉实现方式和懒汉实现方式

[洗完的例子]

吃完饭, 立刻洗碗, 这种就是饿汉方式. 因为下一顿吃的时候可以立刻拿着碗就能吃饭.吃完饭, 先把碗放下, 然后下一顿饭用到这个碗了再洗碗,就是懒汉方式.

懒汉方式最核心的思想是 “延时加载”. 从而能够优化服务器的启动速度.

10.4. 饿汉方式实现单例模式

template <typename T>
class Singleton {
	static T data;
public:
	static T* GetInstance() {
		return &data;
	}
};
1
2
3
4
5
6
7
8

只要通过 Singleton 这个包装类来使用 T 对象, 则一个进程中只有一个 T 对象的实例。

10.5. 懒汉方式实现单例模式

template <typename T>
class Singleton {
	static T* inst;
public:
	static T* GetInstance() {
		if (inst == NULL) {
			inst = new T();
		}
		return inst;
	}
};
1
2
3
4
5
6
7
8
9
10
11

存在一个严重的问题, 线程不安全.
第一次调用 GetInstance 的时候, 如果两个线程同时调用, 可能会创建出两份 T 对象的实例.
但是后续再次调用, 就没有问题了.

10.6. 懒汉方式实现单例模式(线程安全版本)

// 懒汉模式, 线程安全
template <typename T>
class Singleton {
	volatile static T* inst; // 需要设置 volatile 关键字, 否则可能被编译器优化.
	static std::mutex lock;
public:
	static T* GetInstance() {
		if (inst == NULL) { // 双重判定空指针, 降低锁冲突的概率, 提高性能.
			lock.lock(); // 使用互斥锁, 保证多线程情况下也只调用一次 new.
			if (inst == NULL) {
				inst = new T();
			}
			lock.unlock();
		}
	return inst;
	}
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

注意事项:

1. 加锁解锁的位置
2. 双重 if 判定, 避免不必要的锁竞争
3. volatile关键字防止过度优化

11. 自旋与阻塞挂起

无论是是自旋，还是挂起等待，都是等待检测就绪的策略！

• 等待方式不同，原因是什么?
  取决于等待的时长。
  如果时间比较久：推荐其它线程挂起等待。
  如果时间比较短：推荐其它线程不要休眠。
  阻塞，挂起，而是不断一直抢占锁，只到申请成功（自旋）。

   #include <pthread.h>

   int pthread_spin_lock(pthread_spinlock_t *lock);
   int pthread_spin_trylock(pthread_spinlock_t *lock);
   int pthread_spin_unlock(pthread_spinlock_t *lock);

   int pthread_spin_init(pthread_spinlock_t *lock, int pshared);
   int pthread_spin_destroy(pthread_spinlock_t *lock);
1
2
3
4
5
6
7
8

12. 读者写者问题

12.1. 什么是读者写者问题

例子：csdn发文章，印刷报纸，杂志，出黑板报…
读者众多，写者较少——读者问题最常见的情况。
有线程向公共资源写入，其他线程从公共资源中读数据——读者问题。

12. 2. 读者问题 vs 生产消费者问题本质区别？

读者 vs 消费者 区别？
消费者：会把数据拿走！
读者：不会，只会拷贝！

12.3. 伪代码

原理上，我们得理解一下——伪代码——模拟实现一下读者的加锁逻辑

int reader_count = 0;
pthread_mutex_t wlock;
pthread_mutex_t rlock;
1
2
3

读者：

lock(&rlock);
if (reader_count == 0)
	lock(&wlock);  // 申请成功：继续运行，不会有任何读者进来！
				   // 申请失败：阻塞
++reader _count;
unlock(&rlock);
// 可以常规的read了

lock(&rlock);
--reader_count;
if(reader_count == 0)
	unlock(&wlock);
unlock(&rlock); 
1
2
3
4
5
6
7
8
9
10
11
12
13

写者：

lock(&wlock);

// 写入操作

unlock(&wlock);
1
2
3
4
5

12.4. 读写锁接口

pthread_rwlock_t:默认就是读者优先的，会有写者饥饿问题。

总结：

本文全面而深入地探讨了多线程同步的多个方面，从基础概念到高级应用，从理论分析到代码实践，为读者提供了一份详尽的多线程编程指南。通过阅读本文，读者不仅能够理解多线程同步的基本原理和方法，还能够学习到如何在实际编程中应用这些知识，解决多线程环境下的各种同步问题。

在多线程编程中，合理地使用同步机制是确保程序正确性和性能的关键。无论是通过条件变量实现线程间的协调，还是通过生产消费模型、线程池等模式优化资源利用，都需要开发者对同步机制有深刻的理解和熟练的运用能力。同时，避免死锁、确保线程安全和可重入性，也是编写高质量多线程程序的重要考虑因素。

最后，本文还特别强调了在多线程环境下，对于单例模式、自旋锁、读者写者问题等特定场景的处理方法，这些都是多线程编程中不可或缺的知识点。希望通过本文的学习和实践，读者能够在面对复杂的多线程编程挑战时，更加从容不迫，游刃有余。