Linux线程（四）

一、线程同步

1.条件变量

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

2.同步概念与竟态条件

• 同步：在保证数据安全的情况下，让线程能够按照某种特定的顺序访问临界资源，从而有效的避免极饿问题，叫同步
• 竞态条件：因为时许问题而导致出现异常，称为竞态条件。

3.条件变量的初始化

方法一：静态初始化

pthread_cond_t cond = PTHREAD_COND_INITIALIZER

方法二：动态初始化

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

• 功能：初始化一个条件变量
• 参数：pthread_cond_t *restrict cond：待初始化的条件变量
  const pthread_condattr_t *restrict attr：指定条件变量的属性，一般填NULL，使用默认属性

4.pthread_cond_destroy函数

 int pthread_cond_destroy(pthread_cond_t *cond);

• 功能：销毁一个条件变量
• 参数：pthread_cond_t *cond：待销毁的条件变量

5.pthread_cond_wait函数

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

• 功能：等待条件的满足

• 参数：pthread_cond_t *restrict cond：要在这个条件变量上进行等待
  pthread_mutex_t *restrict mutex：互斥量

• 先释放锁（必须是原子的）

• 等待条件的满足（必须是原子的）

• 重新获取锁，继续执行

6.pthread_cond_broadcast和pthread_cond_signal函数

int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

• 功能：唤醒等待

例子：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
pthread_cond_t cond;
pthread_mutex_t mutex;
void *r1( void *arg )
{
while ( 1 ){
pthread_cond_wait(&cond, &mutex);
printf("活动\n");
}
} v
oid *r2(void *arg )
{
while ( 1 ) {
pthread_cond_signal(&cond);
sleep(1);
}
} i
nt main( void )
{
pthread_t t1, t2;
pthread_cond_init(&cond, NULL);
pthread_mutex_init(&mutex, NULL);
pthread_create(&t1, NULL, r1, NULL);
pthread_create(&t2, NULL, r2, NULL);
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_mutex_destroy(&mutex);
pthread_cond_destroy(&cond);
}

7.为什么pthread_cond_wait函数需要用到互斥量

• 条件等待是线程同步的一种手段，如果只有一个线程，条件不满足，一直等下去都会不满足，所以必须要有一个线程通过某些操作，改变共享变量，使原来不满足的条件的到满足，并且友好的通知正在等待条件上的线程
• 条件不会无缘无故的突然变得满足了，必然会牵扯到共享数据的变化，所以需要用互斥锁来保护。没有互斥锁就无法安全的获取和修改共享数据
  
• 按照上面的说法，我们设计出如下的代码：先上锁，发现条件不满足，解锁，然后等待在条件变量上不就行了，如下代码：

// 错误的设计
pthread_mutex_lock(&mutex);
while (condition_is_false) {
pthread_mutex_unlock(&mutex);
//解锁之后，等待之前，条件可能已经满足，信号已经发出，但是该信号可能被错过
pthread_cond_wait(&cond);
pthread_mutex_lock(&mutex);
} p
thread_mutex_unlock(&mutex);

• 由于解锁和等待不是原子操作。调用解锁后，pthread_cond_wait之前，如果已经有其他的线程获取到互斥量，摒弃条件满足，发送了信号，那么pthread_cond_wait将会错过这个信号，可能会导致线程永远阻塞在pthread_cond_wait这里。所以解锁操作必须是原子操作
• 进入pthread_cond_wait这个函数后，会去看条件变量是否等于0，如果等于0，就把互斥量置为1，直到pthread_cond_wait返回，把条件变量改成1，把互斥量恢复成原样

8.条件变量使用规范

等待条件的代码：

pthread_mutex_lock(&mutex);
while (条件为假）
pthread_cond_wait(cond, mutex);
修改条件
pthread_mutex_unlock(&mutex);

给条件发送信号代码：

pthread_mutex_lock(&mutex);
设置条件为真
pthread_cond_signal(cond);
pthread_mutex_unlock(&mutex);

二、生产者消费者模型

1.为什要使用生产者消费者模型 ？

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

2.生产者消费者模型的优点

• 解耦
• 支持并发
• 支持忙闲不均问题

3.生产者消费者模型流程图：

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

例子：单生产者单消费者代码：

#include <iostream>
#include <queue>
#include <stdlib.h>
#include <pthread.h>
#define NUM 8
class BlockQueue{
private:
std::queue<int> q;
int cap;
pthread_mutex_t lock;
pthread_cond_t full;
pthread_cond_t empty;
private:
void LockQueue()
{
pthread_mutex_lock(&lock);
} v
oid UnLockQueue()
{
pthread_mutex_unlock(&lock);
} v
oid ProductWait()
{
pthread_cond_wait(&full, &lock);
} v
oid ConsumeWait()
{
pthread_cond_wait(&empty, &lock);
} v
oid NotifyProduct()
{
pthread_cond_signal(&full);
} v
oid NotifyConsume()
{
pthread_cond_signal(&empty);
} b
ool IsEmpty()
{
return ( q.size() == 0 ? true : false );
} b
ool IsFull()
{
return ( q.size() == cap ? true : false );
}
public:
BlockQueue(int _cap = NUM):cap(_cap)
{
pthread_mutex_init(&lock, NULL);
pthread_cond_init(&full, NULL);
pthread_cond_init(&empty, NULL);
} v
oid PushData(const int &data)
{
LockQueue();
while(IsFull()){
NotifyConsume();
std::cout << "queue full, notify consume data, product stop." << std::endl;
ProductWait();
} q
.push(data);
// NotifyConsume();
UnLockQueue();
} v
oid PopData(int &data)
{
LockQueue();
while(IsEmpty()){
NotifyProduct();
std::cout << "queue empty, notify product data, consume stop." <<
std::endl;
ConsumeWait();
} d
ata = q.front();
q.pop();
// NotifyProduct();
UnLockQueue();
} ~
BlockQueue()
{
pthread_mutex_destroy(&lock);
pthread_cond_destroy(&full);
pthread_cond_destroy(&empty);
}
};
void *consumer(void *arg)
{
BlockQueue *bqp = (BlockQueue*)arg;
int data;
for( ; ; ){
bqp->PopData(data);
std::cout << "Consume data done : " << data << std::endl;
}
} /
/more faster
void *producter(void *arg)
{
BlockQueue *bqp = (BlockQueue*)arg;
srand((unsigned long)time(NULL));
for( ; ; ){
int data = rand() % 1024;
bqp->PushData(data);
std::cout << "Prodoct data done: " << data << std::endl;
// sleep(1);
}
}
i
nt main()
{
BlockQueue bq;
pthread_t c,p;
pthread_create(&c, NULL, consumer, (void*)&bq);
pthread_create(&p, NULL, producter, (void*)&bq);
pthread_join(c, NULL);
pthread_join(p, NULL);
return 0;
}