POSIX信號量——環(huán)形buf實現(xiàn)生產者消費者模型

system v和posix版本信號量的接口函數(shù)

本篇介紹POSIX版本的Semaphore（信號量）

Mutex變量是非0即1的,可看作一種資源的可用數(shù)量,初始化時Mutex是1,表示有一個可用資源,加鎖時獲得該資源,將Mutex減到0,表示不再有可用資源,解鎖時釋放該資源,將Mutex重新加到1,表示又有了一個可用資源。

信號量(Semaphore)和Mutex類似,表示可用資源的數(shù)量,和Mutex不同的是這個數(shù)量可以大于1。如果信號量描述的資源數(shù)目是1時，此時的信號量和互斥鎖相同！

POSIX semaphore庫函數(shù)，這種信號量不僅可用于同一進程的線程間同步,也可用于不同進程間的同步。

有了互斥量和條件變量還提供信號量的原因 是：信號量的主要目的是提供一種進程間同步的方式。這種同步的進程可以共享也可以不共享內存區(qū)。雖然信號量的意圖在于進程間的同步，互斥量和條件變量的意圖在于線程間同步，但信號量也可用于線程間同步，互斥量和條件變量也可通過共享內存區(qū)進行進程間同步。 但應該根據(jù)具體應用考慮到效率和易用性進行具體的選擇。

POSIX版本信號量分為有名信號量和無名信號量

有名信號量函數(shù)：

sem_open 用于創(chuàng)建或打開一個信號量，信號量是通過 name 參數(shù)即信號量的名字來進行標識的。

sem_close 用于關閉打開的信號量。當一個進程終止時，內核對其上仍然打開的所有有名信號量自動執(zhí)行這個操作。

 POSIX 有名信號量是隨內核持續(xù)的。

sem_unlink 用于將有名信號量立刻從系統(tǒng)中刪除，但信號量的銷毀是在所有進程都關閉信號量的時候。

無名信號量有關函數(shù)：

以下代碼是用無名信號量實現(xiàn)的生產者消費者模型

單生產者單消費者模型：

代碼說明：

1.push()和pop()的數(shù)據(jù)(datatype)本篇用的是基本類型,如果是自定義類型的話，需要實現(xiàn)賦值運算符的重載

2.數(shù)組實際上是線性的，內存中并沒有環(huán)形數(shù)組，我們定義了一個固定大小的數(shù)組，當數(shù)組的最后一個元素也被填上數(shù)據(jù)時，檢查數(shù)組的第一個元素（下標為0的元素）是否已經被消費者讀過，如果已經讀過，那么生產者就可以繼續(xù)放數(shù)據(jù)，當數(shù)組滿時（即數(shù)組中的元素一個也沒有被消費者讀），生產者會等待。

3.sem_init()初始化兩個信號量sem_p(控制生產者)和sem_c(控制消費者),pshared為0時，表示信號量用于同一進程的線程間同步

 sem_destroy()使兩個信號量回到初始化前的狀態(tài)

 sem_wait() 可以獲得資源，相當于P操作，把給定信號量減一

 sem_post() 可以釋放資源，相當于V操作，進行加一操作

 當信號量的值為0時，sem_wait()會將進程掛起等待，sem_trywait()不會將進程掛起

4.多生產者和多消費者模型則需要加鎖，其實加用兩把不同的鎖實現(xiàn)效率更高，可以使線程并發(fā)執(zhí)行，而我在下面代碼中指使用了一把鎖，在下面代碼中我把它標記了出來

代碼實現(xiàn)：

ring.cpp

#include <iostream>
#include <stdlib.h>
#include <semaphore.h>
#include <pthread.h>
using namespace std;

#define SEM_PRO 20
#define SEM_CON 0
#define SIZE SEM_PRO

typedef int datatype; 

datatype ring[SIZE];//數(shù)組的定義
datatype pro,con;

sem_t sem_p;//product
sem_t sem_c;//consumer


void init_ring(datatype (*ring)[SIZE])
{
    pro=0;
    con=0;
 }

datatype& push(datatype &data,int index)
{
    ring[pro++]=data;
    datatype tmp=ring[pro-1];
    pro%=SIZE;
    return tmp;
}

datatype& pop(int index)
{
    con++;
    datatype tmp=ring[con-1];
    con%=SIZE;
    return tmp;
}

void* product(void* arg)
{
    while(1){
        datatype data=rand()%50;
        sem_wait(&sem_p);
        datatype val=push(data,pro);
        cout<<"product done...,val is:"<<val<<endl;
        sem_post(&sem_c);
        sleep(1);
    }
}

void* consumer(void* arg)
{
    while(1){
        sem_wait(&sem_c);
        datatype val=pop(con);
        cout<<"consumer done...,val is:"<<val<<endl;
        sem_post(&sem_p);
        sleep(8);
    }
}

int main()
{
    init_ring(&ring);

    sem_init(&sem_p,0,SEM_PRO);
    sem_init(&sem_c,0,0);

    pthread_t tid1,tid2;
    pthread_create(&tid1,NULL,product,NULL);
    pthread_create(&tid2,NULL,consumer,NULL);

    sem_destroy(&sem_p);
    sem_destroy(&sem_c);

    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);

    return 0;
}

Makefile

ring:ring.cpp
    g++ -o $@ $^ -lpthread

.PHONY:clean
clean:
    rm -f ring

下面兩次運行生產者和消費者的速度有所變化，導致運行結果不同

第一次運行結果：

第二次運行結果：

多生產者多消費者模型：

#include <iostream>
#include <stdlib.h>
#include <semaphore.h>
#include <pthread.h>
using namespace std;

#define SEM_PRO 20
#define SEM_CON 0
#define SIZE SEM_PRO

typedef int datatype;

datatype ring[SIZE];
datatype pro,con;

sem_t sem_p;//product
sem_t sem_c;//consumer

void init_ring(datatype (*ring)[SIZE])
{
    pro=0;
    con=0;
}

datatype& push(datatype &data,int index)
{
    ring[pro++]=data;
    datatype tmp=ring[pro-1];
    pro%=SIZE;
    return tmp;
}

datatype& pop(int index)
{
    con++;
    datatype tmp=ring[con-1];
    con%=SIZE;
    return tmp;
}

void* product(void* arg)
{
     while(1){
        int value;
        datatype data=rand()%50;
        sem_wait(&sem_p);//申請資源，進行減一操作
        pthread_mutex_lock(&lock);
        datatype val=push(data,pro);//往buf里push數(shù)據(jù)
        sem_getvalue(&sem_p,&value);//得到此時信號量sem_p的值，即還有幾個空格可>以使用
        pthread_mutex_unlock(&lock);
        cout<<"product"<<(int)arg<<" done...,val is:"<<val<<endl;
        cout<<"sem_p is:"<<value<<",";
        pthread_mutex_lock(&lock);//**********************************應加不同的鎖
        sem_post(&sem_c);//釋放資源
        sem_getvalue(&sem_c,&value);//得到此時信號量sem_c的值，即有多少數(shù)據(jù)可以>取
        pthread_mutex_unlock(&lock);
        cout<<"sem_c is:"<<value<<endl;

        sleep(5);
    }
}

void* consumer(void* arg)
{
     while(1){
        int value;
        sem_wait(&sem_c);
        pthread_mutex_lock(&lock);
        datatype val=pop(con);
        sem_getvalue(&sem_c,&value);
        pthread_mutex_unlock(&lock);
        cout<<"consumer"<<(int)arg<<" done...,val is:"<<val<<endl;
        cout<<"sem_c is:"<<value<<",";
        pthread_mutex_lock(&lock);//**********************************應加不同的鎖
        sem_post(&sem_p);
        sem_getvalue(&sem_p,&value);
        pthread_mutex_unlock(&lock);
        cout<<"sem_p is:"<<value<<endl;
        sleep(1);
    }
}

int main()
{
    init_ring(&ring);

    sem_init(&sem_p,0,SEM_PRO);
    sem_init(&sem_c,0,0);

    pthread_t tid1,tid2,tid3;
    pthread_create(&tid1,NULL,product,(void*)1);
    pthread_create(&tid2,NULL,product,(void*)2);
    pthread_create(&tid3,NULL,product,(void*)3);

    pthread_t tid4,tid5;
    pthread_create(&tid4,NULL,consumer,(void*)4);
    pthread_create(&tid5,NULL,consumer,(void*)5);

    sem_destroy(&sem_p);
    sem_destroy(&sem_c);

    pthread_join(tid1,NULL);
    pthread_join(tid2,NULL);
    pthread_join(tid3,NULL);
    pthread_join(tid4,NULL);
    pthread_join(tid5,NULL);

    return 0;
}

運行結果：

很多時候信號量和互斥量，條件變量三者都可以再某種應用中使用，那這三者的差異有哪些呢，下面列出了這三者之間的差異 ：

• 互斥量必須由給它上鎖的線程解鎖。而信號量不需要由等待它的線程進行掛出，可以在其他進程進行掛出操作。

• 互斥量要么被鎖住，要么是解開狀態(tài)，只有這兩種狀態(tài)。而信號量的值可以支持多個進程成功進行 wait 操作。

• 信號量的掛出操作總是被記住，因為信號量有一個計數(shù)值，掛出操作總會將該計數(shù)值加 1 ，然而當向條件變量發(fā)送一個信號時，如果沒有線程等待在條件變量，那么該信號會丟失。