开发者社区> 翡青> 正文

Linux多线程实践(8) --Posix条件变量解决生产者消费者问题

简介: Posix条件变量 int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr); int pt...
+关注继续查看

Posix条件变量

int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);
int pthread_cond_destroy(pthread_cond_t *cond);

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_timedwait(pthread_cond_t *cond,  pthread_mutex_t  *mutex,  const  struct timespec *abstime);

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

    与互斥锁不同,条件变量是用来等待而不是用来上锁的。条件变量用来自动阻塞调用线程, 直到条件变量所要求的情况发生为止。通常条件变量需要和互斥锁同时使用, 利用互斥量保护条件变量;

  条件的检测是在互斥锁的保护下进行的。如果一个条件为假,一个线程自动阻塞,并释放等待状态改变的互斥锁。如果另一个线程改变了条件,它就发送信号给关联的条件变量, 并唤醒一个或多个等待在该条件变量上的线程,这些线程将重新获得互斥锁,重新评价条件。如果将条件变量放到共享内存中, 而两进程可共享读写这段内存,则条件变量可以被用来实现两进程间的线程同步。


条件变量使用规范

1.等待条件代码

pthread_mutex_lock(&mutex);

while (条件为假)
{
    pthread_cond_wait(&cond, &mutex);
}
修改条件

pthread_mutex_unlock(&mutex);

/**解释: 为什么使用while, 而不用if?

Man-Page给出了答案: If a signal is delivered to a thread waiting for a condition variable, upon return from 

the signal handler the thread resumes waiting for the condition variable as if it was not interrupted, or 

it shall return zero due to spurious wakeup.

即是说如果正在等待条件变量的一个线程收到一个信号,从信号处理函数返回的时候线程应该重新等待条件变量就好象没有被中断一样,或者被虚假地唤醒返回0。如果是上述情形,那么其实条件并未被改变,那么此时如果没有继续判断一下条件的真假就继续向下执行的话,修改条件将会出现问题,所以需要使用while 循环再判断一下,如果条件还是为假必须继续等待。

注:在多处理器系统中,pthread_cond_signal 可能会唤醒多个等待条件的线程,这也是一种spurious wakeup。

**/

2.给条件发送信号代码

pthread_mutex_lock(&mutex);

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

pthread_mutex_unlock(&mutex);


条件变量API说明

1.pthread_cond_init

使用条件变量之前要先进行初始化:可以在单个语句中生成和初始化一个条件变量如:

  pthread_cond_t my_condition=PTHREAD_COND_INITIALIZER; //用于进程间线程的通信;

  或用函数pthread_cond_init进行动态初始化;

 

2.pthread_cond_destroy

该函数可以用来摧毁所指定的条件变量,同时将会释放所给它分配的资源。调用该函数的进程并不要求等待在参数所指定的条件变量上;

 

3.pthread_cond_wait && pthread_cond_timedwait

cond_wait原语完成三件事:

  (1)对mutex解锁;

  (2)等待条件, 直到有线程向他发送通知;

  (3)当wait返回时, 再对mutex重新加锁;

第一个参数cond是指向一个条件变量的指针。第二个参数mutex则是对相关的互斥锁的指针。

函数pthread_cond_timedwait函数类型与函数pthread_cond_wait区别在于:timedwait多了一个超时, 超时值制订了我们愿意等待多长时间, 如果达到或是超过所引用的参数*abstime,它将结束阻塞并返回错误ETIME.

//timespec结构如下:
struct timespec
{
    time_t   tv_sec;        /* seconds */
    long     tv_nsec;       /* nanoseconds */
};

注意: 这个时间值是一个绝对数而不是相对数, 例如, 假设愿意等待三秒钟, 那么并不是把3秒钟转换成timespec结构, 而是需要将当前实践加上3分钟再转换成timespec结构, 这个获取当前时间值的函数可以是clock_gettime(我们采用这一个)也可以是gettimeofday.

 

4.pthread_cond_signal && pthread_cond_broadcast

cond_signal原语所完成的操作:

  向第一个等待条件的线程发起通知, 如果没有任何一个线程处于等待条件的状态, 那么这个通知将被忽略;

cond_broadcast:

  向所有等待在该条件上的线程发送通知;

参数cond是一个条件变量的指针。当调用signal时, 一个在相同条件变量上阻塞的线程将被解锁。如果同时有多个线程阻塞,则由调度策略确定接收通知的线程。如果调用broadcast,则将通知阻塞在这个条件变量上的所有线程。一旦被唤醒,线程仍然会要求互斥锁。如果当前没有线程等待通知,则上面两种调用实际上成为一个空操作, 内核会将条件变量的通知忽略(如果参数*cond指向非法地址,则返回值EINVAL);

 

类Condition封装

//Condition类设计
class Condition
{
public:
    Condition(const pthread_mutexattr_t *mutexAttr = NULL,
              const pthread_condattr_t  *condAttr = NULL);
    ~Condition();

    //条件变量函数
    int signal();
    int broadcast();
    int wait();
    int timedwait(int seconds);

    //互斥量函数
    int lock();
    int trylock();
    int unlock();

private:
    pthread_mutex_t m_mutex;
    pthread_cond_t  m_cond;
};
//Condition类实现
Condition::Condition(const pthread_mutexattr_t *mutexAttr,
                     const pthread_condattr_t  *condAttr)
{
    //初始化互斥量
    pthread_mutex_init(&m_mutex, mutexAttr);
    //初始化条件变量
    pthread_cond_init(&m_cond, condAttr);
}
Condition::~Condition()
{
    //销毁互斥量
    pthread_mutex_destroy(&m_mutex);
    //销毁条件变量
    pthread_cond_destroy(&m_cond);
}
int Condition::signal()
{
    return pthread_cond_signal(&m_cond);
}
int Condition::broadcast()
{
    return pthread_cond_broadcast(&m_cond);
}
int Condition::wait()
{
    return pthread_cond_wait(&m_cond, &m_mutex);
}
int Condition::timedwait(int seconds)
{
    //获取当前时间
    struct timespec abstime;
    clock_gettime(CLOCK_REALTIME, &abstime);
    //将当前时间加上需要等待的秒数, 构成绝对时间值
    abstime.tv_sec += seconds;
    return pthread_cond_timedwait(&m_cond, &m_mutex, &abstime);
}

int Condition::lock()
{
    return pthread_mutex_lock(&m_mutex);
}
int Condition::trylock()
{
    return pthread_mutex_trylock(&m_mutex);
}
int Condition::unlock()
{
    return pthread_mutex_unlock(&m_mutex);
}

生产者消费者问题(无界缓冲区)

/** 实现: 我们假设是缓冲区是无界的
说明:生产者可以不停地生产,使用pthread_cond_signal  发出通知的时候,如果此时没有消费者线程在等待条件,那么这个通知将被丢弃,但也不影响整体代码的执行,没有消费者线程在等待,说明产品资源充足,即while 判断失败,不会进入等待状态,直接消费产品(即修改条件)。
**/
const unsigned int PRODUCER_COUNT = 5;	//生产者个数
const unsigned int CONSUMER_COUNT = 3;	//消费者个数

//定义Condition类
Condition cond;
//缓冲区 ~O(∩_∩)O~
int nReady = 0;
//消费者
void *consumer(void *args)
{
    int id = *(int *)args;
    delete (int *)args;
    while (true)
    {
        cond.lock();    //锁定mutex
        while (!(nReady > 0))
        {
            printf("-- thread %d wait...\n", id);
            cond.wait();    //等待条件变量
        }

        printf("** thread %d alive, and consume product %d ...\n", id, nReady);
        -- nReady;  //消费
        printf("   thread %d end consume... \n\n", id);

        cond.unlock();  //解锁mutex
        sleep(1);
    }
    pthread_exit(NULL);
}

//生产者
void *producer(void *args)
{
    int id = *(int *)args;
    delete (int *)args;
    while (true)
    {
        cond.lock();    //锁定mutex

        printf("++ thread %d signal, and produce product %d ...\n", id, nReady+1);
        ++ nReady;      //生产
        cond.signal();  //发送条件变量信号
        printf("   thread %d end produce, signal...\n\n", id);
        cond.unlock();  //解锁mutex
        sleep(1);
    }
    pthread_exit(NULL);
}

int main()
{
    pthread_t thread[PRODUCER_COUNT+CONSUMER_COUNT];

    //首先生成消费者
    for (unsigned int i = 0; i < CONSUMER_COUNT; ++i)
        pthread_create(&thread[i], NULL, consumer, new int(i));
    sleep(1);   //使生产者等待一段时间, 加速消费者等待事件产生
    //然后生成生产者
    for (unsigned int i = 0; i < PRODUCER_COUNT; ++i)
        pthread_create(&thread[CONSUMER_COUNT+i], NULL, producer, new int(i));
    for (unsigned int i = 0; i < PRODUCER_COUNT+CONSUMER_COUNT; ++i)
        pthread_join(thread[i], NULL);
}

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
【Linux】多线程 --- POSIX信号量+懒汉模式的线程池+其他常见锁
【Linux】多线程 --- POSIX信号量+懒汉模式的线程池+其他常见锁
11 0
【Linux】POSIX信号量
POSIX和System V都是可移植的操作系统接口标准,它们都定义了操作系统应该为应用程序提供的接口标准。
10 0
一文读懂Linux多线程中互斥锁、读写锁、自旋锁、条件变量、信号量
Hello、Hello大家好,我是木荣,今天我们继续来聊一聊Linux中多线程编程中的重要知识点,详细谈谈多线程中同步和互斥机制。
8407 0
【Linux线程同步专题】三、条件变量
【Linux线程同步专题】三、条件变量
33 0
Linux网络编程必备的POSIX API的细节
Linux网络编程必备的POSIX API的细节
86 0
Linux系统中线程同步方式中的条件变量操作方法
大家好,今天主要和大家聊一聊,如何使用Linux中线程同步方式中的条件变量。
23 0
Linux信号量:POSIX标准接口、实现生产者与消费者模型
Linux信号量:POSIX标准接口、实现生产者与消费者模型
64 0
Linux系统编程-(pthread)线程通信(条件变量)
条件变量是线程可用的一种同步机制,条件变量给多个线程提供了一个回合的场所,条件变量和互斥量一起使用,允许线程以无竞争的方式等待特定的条件发生。 条件变量本身是由互斥体保护的,线程在改变条件状态之前必须首先锁住互斥量,其他线程在获取互斥量之前就不会觉察到这种变化,因为互斥量必须锁定之后才改变条件。
86 0
Linux Qt使用POSIX多线程条件变量、互斥锁(量)
Linux Qt使用POSIX多线程条件变量、互斥锁(量)今天团建,但是文章也要写。酒要喝好,文要写美,方为我辈程序员的全才之路。嘎嘎 之前一直在看POSIX的多线程编程,上个周末结合自己的理解,写了一个基于Qt的用条件变量同步线程的例子。
1259 0
+关注
翡青
阿里巴巴高级研发工程师
文章
问答
视频
文章排行榜
最热
最新
相关电子书
更多
ECS运维指南 之 Linux系统诊断
立即下载
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载
相关实验场景
更多
相关镜像