Linux之线程进阶(一)

同进程内多个线程共享地址空间、共享进程资源

↓

多个线程同时访问同一份共享资源

↓

典型现象：多线程同时打印到显示器（stdout，fd=1），输出乱序、文字穿插干扰

↓

本质问题：并发访问共享资源带来的数据不一致问题

一个现象

#include <iostream>
#include "Thread.hpp"
#include <unistd.h>
#include <string>
#include <vector>
using namespace std;

int tickets = 1000;

void GetTicket()
{
char name[64];
pthread_getname_np(pthread_self(), name, sizeof(name));
while (1)
{
if (tickets > 0)
{
usleep(1000);
printf("%s sells ticket:%d\\n", name, tickets);
tickets–;
}
else
{
break;
}
}
}

int main()
{

ThreadModul::Thread t1(GetTicket);
ThreadModul::Thread t2(GetTicket);
ThreadModul::Thread t3(GetTicket);
ThreadModul::Thread t4(GetTicket);
t1.Start();
t2.Start();
t3.Start();
t4.Start();
sleep(10);
t1.Join();
t2.Join();
t3.Join();
t4.Join();
return 0;
}

初始 tickets = 1

 重点：两个线程都已经通过了 if(tickets>0) 的校验！ 此时票还没减！

于是就出现：票已经 0 了，还能继续卖出，出现负数 你截图里 -1 -2 就是多轮这种叠加导致的

同步

• VIP 自习室 = 临界资源（同一时间只允许 1 个人用）
• 人 = 线程
• 自习的过程 = 临界区
• 钥匙 = 互斥锁（pthread_mutex）

规则：

✅ 线程同步定义：在临界资源安全的前提下，让访问临界资源具有一定的顺序性！

💡区分：

• 互斥：保证同一时刻，只有一个线程进入临界区（解决安全问题，最基础）
• 同步：在互斥基础上，控制线程访问的先后顺序（解决顺序问题）

生产者消费者模型

这是多线程协同的一种模式,提高协作效率,本质是一种通信工具

3 种关系：生产者和生产者，消费者和消费者，生产者和消费者

• 生产者 ↔ 生产者：互斥
• 消费者 ↔ 消费者：互斥
• 生产者 ↔ 消费者：同步 + 互斥

2 种角色：生产者线程、消费者线程

1 一个交易场所：超市（缓冲区 / 内存空间），一般用队列、环形队列这类数据结构实现

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间 不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔 给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区， 平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

✨生产者消费者模型的优势

理解条件变量

条件变量，就是让线程「等某个条件满足再干活」，不满足就休眠，不瞎循环空耗 CPU；必须搭配互斥锁一起使用！

互斥锁：管能不能进临界区（互斥） 条件变量：管什么时候可以干活（等待 + 唤醒，同步）

例子

宿舍饮水机：

• 互斥锁 = 饮水机钥匙，同一时间只能 1 个人接水
• 条件变量 = “有水吗” 这个信号
• 你来接水，拿到钥匙，发现没水 → 不一直反复拧钥匙空等，直接休眠，释放钥匙（pthread_cond_wait自动解锁 + 休眠）
• 送水工（另一个线程）换水完成，发信号唤醒等待的人（pthread_cond_signal）
• 你被唤醒后，自动重新拿到锁，再检查有没有水，有水再接水

代码

void* Print(void* args)
{
string name=static_cast<const char*>(args);
while(1)
{
cout<<"我是新线程:"<<name<<endl;

}
return nullptr;
}

int main()
{
pthread_t tids[4];
for(int i=0;i<4;i++)
{
char *name=new char[64];

snprintf(name,64,"thread-%d",i+1);
pthread_create(tids+i,nullptr,Print,name);
}
for(int i=0;i<4;i++)
{
void *retval;
pthread_join(tids[i],&retval);
}
return 0;
}

上锁

阻塞等待

pthread_mutex_t gmutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t gcond=PTHREAD_COND_INITIALIZER;

void* Print(void* args)
{
string name=static_cast<const char*>(args);
while(1)
{
pthread_mutex_lock(&gmutex); // 1.抢锁
cout<<"我是新线程:"<<name<<endl; // 2.打印
pthread_cond_wait(&gcond,&gmutex); // 3.【重点】释放锁 + 阻塞休眠，等唤醒信号
pthread_mutex_unlock(&gmutex); // 4.被唤醒后，抢到锁才走到这里解锁
}
return nullptr;
}

唤醒

#include <iostream>
#include <unistd.h>
#include <string>
#include <vector>
using namespace std;

pthread_mutex_t gmutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t gcond=PTHREAD_COND_INITIALIZER;

void* Print(void* args)
{
string name=static_cast<const char*>(args);
while(1)
{
pthread_mutex_lock(&gmutex);
cout<<"我是新线程:"<<name<<endl;
pthread_cond_wait(&gcond,&gmutex);
pthread_mutex_unlock(&gmutex);
sleep(1);
}
return nullptr;
}

int main()
{
pthread_t tids[4];
for(int i=0;i<4;i++)
{
char *name=new char[64];

snprintf(name,64,"thread-%d",i+1);
pthread_create(tids+i,nullptr,Print,name);
}

while(1)
pthread_cond_signal(&gcond);
for(int i=0;i<4;i++)
{
void *retval;
pthread_join(tids[i],&retval);
}
return 0;
}

阻塞

整体结构说明

ProdurorRoutine

ConsumerRoutine

接下来我们来实现封装

构造和析构

Pop

线程为什么会在临界区内部等待？

疑惑：临界区是要快速执行、尽快解锁，怎么还在里面阻塞等待？ 核心一句话：判断临界资源是否就绪这件事本身，就属于临界区操作！ 我们要判断阻塞队列_bq是否为空，队列是多线程共享的临界资源，读它也要受互斥锁保护，所以这个判断逻辑，必须放在pthread_mutex_lock之后、持有锁的临界区内。 如果我们先解锁、再判断队列状态，就会出现时序竞态：生产者修改队列的同时消费者在读队列，造成数据错乱、信号丢失。

 顺序不能乱：加锁 → 判断资源状态 → 资源不就绪就 wait 等待

为什么pthread_cond_wait一定要传入互斥锁？

pthread_cond_wait不是单纯休眠，它和传入的 mutex 绑定，内部自动完成一套原子操作，分为 3 步：

Enqueue

和pop类似,但是面对的是productor

问题

但是当队列满了,如何唤醒consumer

队列快要空了,如何唤醒productor

目前我们先让这个生产一个,消费一个

生产者较快

消费者更快

细节问题

伪唤醒

可能会有过量唤醒信息,函数调用失败等等

POSIX 标准允许：没有任何线程调用 signal，cond_wait 也可能莫名唤醒。

• 如果用if(_bq.empty())：只判断一次。一旦伪唤醒，不会二次检查，直接执行 pop，空队列取数据，程序异常
• 如果用while(_bq.empty())：唤醒后循环复检条件，队列依旧为空就继续 wait，保证安全

规范写法：永远 while

带高低水位线的有界阻塞队列

void Enqueue(T &in)
{
pthread_mutex_lock(&_mutex);
while (_bq.size() == _cap)
{
pthread_cond_wait(&_productor_cond, &_mutex);
}
_bq.push(in);
if (_bq.size() >= _highline)
{
pthread_cond_signal(&_consumer_cond);
}
pthread_mutex_unlock(&_mutex);
}
void Pop(T *out)
{
pthread_mutex_lock(&_mutex);
while (_bq.empty())
{
pthread_cond_wait(&_consumer_cond, &_mutex);
}
*out = _bq.front();
_bq.pop();
if (_bq.size() <= _lowline)
{
pthread_cond_signal(&_productor_cond);
}
pthread_mutex_unlock(&_mutex);
}

统计休眠线程数量的唤醒策略

void Enqueue(T &in)
{
pthread_mutex_lock(&_mutex);
while (_bq.size() == _cap)
{
_sleep_pro_num++;
pthread_cond_wait(&_productor_cond, &_mutex);
_sleep_pro_num–;
}
_bq.push(in);
// if (_bq.size() >= _highline)
// {
// pthread_cond_signal(&_consumer_cond);
// }

if (_sleep_con_num > 0)
{
pthread_cond_signal(&_consumer_cond);
}
pthread_mutex_unlock(&_mutex);
}
void Pop(T *out)

多线程多任务

以上都是单线程单任务,那我们能不能实现多线程多任务???

#include <iostream>
#include <unistd.h>
#include <string>
#include <vector>
using namespace std;
#include "BlockQueue.hpp"

int num = 0;
pthread_mutex_t mutex;

int GetNumber()
{
pthread_mutex_lock(&mutex);
num++;
pthread_mutex_unlock(&mutex);
return num;
}

void *ConsumerRoutine(void *args)
{
int number = GetNumber();
BlockQueue<int> *bp = static_cast<BlockQueue<int> *>(args);
while (1)
{
int data;
bp->Pop(&data);
cout << "con–" << number << " ";

cout << "我是消费者,消费了–" << data << endl;
}
}
void *ProdutorRoutine(void *args)
{
BlockQueue<int> *bp = static_cast<BlockQueue<int> *>(args);
int data = 10;
int number = GetNumber();
while (1)
{
sleep(1);
bp->Enqueue(data);
cout << "pro–" << number << " ";

cout << "我是生产者,生产了–" << data++ << endl;
}
}
int main()
{

BlockQueue<int> *bp = new BlockQueue<int>();
pthread_t c[3], p[2];
pthread_create(c, nullptr, ConsumerRoutine, bp);
pthread_create(c + 1, nullptr, ConsumerRoutine, bp);
pthread_create(c + 2, nullptr, ConsumerRoutine, bp);
pthread_create(p, nullptr, ProdutorRoutine, bp);
pthread_create(p + 1, nullptr, ProdutorRoutine, bp);

pthread_join(c[0], nullptr);
pthread_join(c[1], nullptr);
pthread_join(c[2], nullptr);
pthread_join(p[0], nullptr);
pthread_join(p[1], nullptr);
return 0;
}