LINUX学习:生产者&消费者模型复习

浏览数：18 / 时间：2015年06月20日

回顾一下生产者消费者模型。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <unistd.h>
#include <sys/types.h>
#include <pthread.h>
#define ERR_EXIT(m)     do {         perror(m);        exit(EXIT_FAILURE);    }while(0)
#define CONSUMERS_COUNT 1
#define PRODUCERS_COUNT 2
#define BUFFSIZE 10

int g_buffer[BUFFSIZE];

unsigned short in = 0;
unsigned short out = 0;
unsigned short product_id = 0;
unsigned short consume_id = 0;

sem_t g_sen_full;
sem_t g_sen_empty;
pthread_mutex_t g_mutex;

pthread_t g_thread[CONSUMERS_COUNT+PRODUCERS_COUNT];

void* consume(void* arg)
{
    int i;
    while(1)
    {
        sem_wait(&g_sen_empty);
        pthread_mutex_lock(&g_mutex);
        for (i = 0; i < BUFFSIZE; ++i)
        {
            /* code */
            printf("%02d\n", i);
            if(g_buffer[i] == -1)
                printf("%s", "null");
            else
                printf("%d\n", g_buffer[i]);
            if(i == out)
                printf("\t<--consume\n");
            printf("\n");
        }
        consume_id = g_buffer[out];
        printf("begin consume product %d\n", consume_id);
        g_buffer[out] = -1;
        out = (out + 1) % BUFFSIZE;
        pthread_mutex_unlock(&g_mutex);
        sem_post(&g_sen_full);
        sleep(1);
    }
    return NULL;
}
void* produce(void *arg)
{
    int num = (int)arg;
    int i;
    while(1)
    {
        printf("%d waiting buffer_full\n", num);
        sem_wait(&g_sen_full);
        pthread_mutex_lock(&g_mutex);
        for (i = 0; i < BUFFSIZE; ++i)
        {
            /* code */
            printf("%02d\n", i);
            if(g_buffer[i] == -1)
                printf("%s", "null");
            else
                printf("%d\n", g_buffer[i]);
            if(i == in)
                printf("\t<--produce\n");
            printf("\n");
        }
        printf("begin produce product %d\n", product_id);
        g_buffer[in] = product_id;
        in = (in +1) % BUFFSIZE;
        printf("end produce product %d\n", product_id++);
        pthread_mutex_unlock(&g_mutex);
        sem_post(&g_sen_empty);
    }
    return NULL;
}
int main(int argc, const char *argv[])
{
    int i;
    for (i = 0; i < BUFFSIZE; ++i)
    {
        /* code */
        g_buffer[i] = -1;
    }
    // 初始化信号量
    sem_init(&g_sen_full, 0, BUFFSIZE);
    sem_init(&g_sen_empty, 0, 0);
    // 初始化锁
    pthread_mutex_init(&g_mutex, NULL);

    for (i = 0; i < CONSUMERS_COUNT; ++i)
    {
        /* code */
        pthread_create(&g_thread[i], NULL, consume, (void*)i);
    }
        for (i = 0; i < PRODUCERS_COUNT; ++i)
    {
        /* code */
        pthread_create(&g_thread[CONSUMERS_COUNT+i], NULL, produce, (void*)i);
    }
        for (i = 0; i < CONSUMERS_COUNT+PRODUCERS_COUNT; ++i)
    {
        /* code */
        pthread_join(g_thread[i], NULL);
    }

    sem_destroy(&g_sen_empty);

    sem_destroy(&g_sen_full);
    pthread_mutex_destroy(&g_mutex);  
    return 0;
}

东莞东方

再写一个条件变量的版本。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <unistd.h>
#include <sys/types.h>
#include <pthread.h>
#define ERR_EXIT(m)     do {         perror(m);        exit(EXIT_FAILURE);    }while(0)
#define CONSUMERS_COUNT 2
#define PRODUCERS_COUNT 1
#define BUFFSIZE 10



unsigned short in = 0;
unsigned short out = 0;
unsigned short product_id = 0;
unsigned short consume_id = 0;

sem_t g_sen_full;
sem_t g_sen_empty;
pthread_mutex_t g_mutex;
pthread_cond_t g_cond;
pthread_t g_thread[CONSUMERS_COUNT+PRODUCERS_COUNT];

int nready = 0;

void* consume(void* arg)
{
    int num = (int)arg;
    while(1)
    {
        pthread_mutex_lock(&g_mutex);
        while(nready==0)
        {
            printf("begin wait a condtion\n");
             pthread_cond_wait(&g_cond, &g_mutex);
        }

        printf("no.%dend of wait a condtion ....\n", num);
        --nready;
        printf("消费者%d消耗一个产品\n", num);
        pthread_mutex_unlock(&g_mutex);

    }
    return NULL;
}
void* produce(void *arg)
{
    int num = (int)arg;
    while(1)
    {
        pthread_mutex_lock(&g_mutex);
        ++nready;
        printf("生产者%d生产了一个产品并发送signal..\n", num);
        pthread_cond_signal(&g_cond);
        pthread_mutex_unlock(&g_mutex);
        sleep(5);
    }
    return NULL;
}
int main(int argc, const char *argv[])
{
    int i;
    pthread_mutex_init(&g_mutex, NULL);
    pthread_cond_init(&g_cond, NULL);

    for (i = 0; i < CONSUMERS_COUNT; ++i)
    {
        /* code */
        pthread_create(&g_thread[i], NULL, produce, (void*)i);
    }
    for (i = 0; i < PRODUCERS_COUNT; ++i)
    {
        /* code */
        pthread_create(&g_thread[i+CONSUMERS_COUNT], NULL, consume, (void*)i);
    }
    for (i = 0; i < CONSUMERS_COUNT+PRODUCERS_COUNT; ++i)
    {
        /* code */
        pthread_join(g_thread[i], NULL);
    }
    pthread_mutex_destroy(&g_mutex);
    pthread_cond_destroy(&g_cond);
    return 0;
}