使用信號量實(shí)現(xiàn)了多個生產(chǎn)者，多個消費(fèi)者，多個緩沖區(qū)的消費(fèi)者和生產(chǎn)者的解決方案。

模擬場景： 15個生產(chǎn)者，10個消費(fèi)者，5個緩沖區(qū)，緩沖區(qū)數(shù)據(jù)結(jié)構(gòu)包含數(shù)據(jù)、生產(chǎn)線程tid、寫入時間。

編譯時使用： 

gcc -o kk kk.c -lpthread

#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <unistd.h>  //getpid()
#include <semaphore.h> 
#include <fcntl.h>
#include <errno.h>
#include <sys/syscall.h>   //gettid() thread
#include<time.h>
#include <string.h>      //for memset
//創(chuàng)建模式權(quán)限
#define FILE_MODE  (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)
#define BUFFER 5 // 緩沖區(qū)數(shù)量
#define PRO_NO 30 // PRODUCING NO
#define OVER ( - 1)
#define PSLEEP 10000  // 
#define CSLEEP 10000  // 

#define PRODUCER 15  // 生產(chǎn)者線程數(shù)量
#define CONSUMER 10  // 消費(fèi)者線程數(shù)量
#define gettid() syscall(SYS_gettid)   //獲取線程ID

    sem_t empty,full;//同步信號量
    pthread_mutex_t mutex;//互斥信號量

struct prodcons
{// 緩沖區(qū)相關(guān)數(shù)據(jù)結(jié)構(gòu)
    int buf[BUFFER]; /* 實(shí)際數(shù)據(jù)存放的數(shù)組*/
    int tid[BUFFER];            //生產(chǎn)進(jìn)程PID   getpid() 頭文件
    char *time[BUFFER];         //寫入時間
    int readpos, writepos; /* 讀寫指針*/
};
struct prodcons buffer;

/* 初始化緩沖區(qū)結(jié)構(gòu) */
void init(struct prodcons *b)
{
    b->readpos = 0;
    b->writepos = 0;
}
int producer_id=0,consumer_id=0;//生產(chǎn)者消費(fèi)者ID


void *Producer()//生產(chǎn)者函數(shù)
{
int n=0;
time_t timep;
time(&timep);
char *tem;
int nn=PRO_NO;
   while(nn--)
   {

     sleep(3);
     sem_wait(&empty);
     pthread_mutex_lock(&mutex);
     
    buffer.buf[buffer.writepos] = n;
    buffer.tid[buffer.writepos] = gettid();  //threadid
    tem = asctime(gmtime(&timep));
    buffer.time[buffer.writepos] = tem; //time
    printf("data:%d ,tid: %ld ,time_in: %s --->\n", n,gettid(),tem);
        
    buffer.writepos++;
    if (buffer.writepos >= BUFFER)
        buffer.writepos = 0;
     pthread_mutex_unlock(&mutex);
     sem_post(&full);
     n++;
   }
   
}
void *Consumer()//消費(fèi)者函數(shù)
{
    int d;
int gettid;
char *tt;
   while(1)
   {
     sleep(3);
     sem_wait(&full);
     pthread_mutex_lock(&mutex);
    /* 讀數(shù)據(jù),移動讀指針*/
    d = buffer.buf[buffer.readpos];
    gettid = buffer.tid[buffer.readpos];
    tt = buffer.time[buffer.readpos];
    //usleep(CSLEEP);
    buffer.readpos++;
    if (buffer.readpos >= BUFFER)
        buffer.readpos = 0;
    printf("--->data:%d ,tid: %d ,time_in: %s \n", d,gettid,tt);
    if (d == OVER)
        break;
     
     pthread_mutex_unlock(&mutex);
     sem_post(&empty);
   }
}

int main()
{
init(&buffer);
   int rthread[18],i;
   pthread_t producer[PRODUCER];//生產(chǎn)者
   pthread_t consumer[CONSUMER];//消費(fèi)者

   int sinit1=sem_init(&empty,0,BUFFER);//初始化同步信號量
   int sinit2=sem_init(&full,0,0);
   int minit =pthread_mutex_init(&mutex,NULL);//初始化互斥信號量
   if(sinit1 && sinit2)
   {
     printf("sem initialize failed /n");
     exit(1);
   }
   if(minit)
   {
     printf("sem initialize failed /n");
     exit(1);
   }
   for(i=0;i<PRODUCER;i++)//創(chuàng)建生產(chǎn)者線程
   {
      rthread[i]=pthread_create(&producer[i], NULL, Producer, NULL);
      if(rthread[i])
      {
          printf("producer %d create failed /n", i);
          exit(1);
      }
   }
   for(i=0;i<CONSUMER;i++)//創(chuàng)建消費(fèi)者線程
   {
      rthread[i]=pthread_create(&consumer[i], NULL, Consumer,NULL);
      if(rthread[i])
      {
          printf("consumer %d create failed /n", i);
          exit(1);
      }
   }
   for(i=0;i<PRODUCER;i++)//銷毀生產(chǎn)者線程
   {
      pthread_join(producer[i],NULL);
   }
   for(i=0;i<CONSUMER;i++)//銷毀生產(chǎn)者線程
   {
      pthread_join(consumer[i],NULL);
   }
   exit(0);
}