多线程编程之：实验内容——“生产者消费者”实验

（3）编写代码

本实验的代码中采用的有界缓冲区拥有3个单元，每个单元为5个字节。为了尽量体现每个信号量的意义，在程序中生产过程和消费过程是随机（采取0～5s的随机时间间隔）进行的，而且生产者的速度比消费者的速度平均快两倍左右（这种关系可以相反）。生产者一次生产一个单元的产品（放入“hello”字符串），消费者一次消费一个单元的产品。

/*producer-customer.c*/

#includestdio.h>

#includestdlib.h>

#includeunistd.h>

#includefcntl.h>

#includepthread.h>

#includeerrno.h>

#includesemaphore.h>

#includesys/ipc.h>

#defineMYFIFOmyfifo/*缓冲区有名管道的名字*/

#defineBUFFER_SIZE3/*缓冲区的单元数*/

#defineUNIT_SIZE5/*每个单元的大小*/

#defineRUN_TIME30/*运行时间*/

#defineDELAY_TIME_LEVELS5.0/*周期的最大值*/

intfd;

time_tend_time;

sem_tmutex,full,avail;/*3个信号量*/

/*生产者线程*/

void*producer(void*arg)

{

intreal_write;

intdelay_time=0;

while(time(NULL)end_time)

{

delay_time=(int)(rand()*DELAY_TIME_LEVELS/(RAND_MAX)/2.0)+1;

sleep(delay_time);

/*P操作信号量avail和mutex*/

sem_wait(avail);

sem_wait(mutex);

printf(nProducer:delay=%dn,delay_time);

/*生产者写入数据*/

if((real_write=write(fd,hello,UNIT_SIZE))==-1)

{

if(errno==EAGAIN)

{

printf(TheFIFOhasnotbeenreadyet.Pleasetrylatern);

}

}

else

{

printf(Write%dtotheFIFOn,real_write);

}

/*V操作信号量full和mutex*/

sem_post(full);

sem_post(mutex);

}

pthread_exit(NULL);

}

/*消费者线程*/

void*customer(void*arg)

{

unsignedcharread_buffer[UNIT_SIZE];

intreal_read;

intdelay_time;

while(time(NULL)end_time)

{

delay_time=(int)(rand()*DELAY_TIME_LEVELS/(RAND_MAX))+1;

sleep(delay_time);

/*P操作信号量full和mutex*/

sem_wait(full);

sem_wait(mutex);

memset(read_buffer,0,UNIT_SIZE);

printf(nCustomer:delay=%dn,delay_time);

if((real_read=read(fd,read_buffer,UNIT_SIZE))==-1)

{

if(errno==EAGAIN)

{

printf(Nodatayetn);

}

}

printf(Read%sfromFIFOn,read_buffer);

/*V操作信号量avail和mutex*/

sem_post(avail);

sem_post(mutex);

}

pthread_exit(NULL);

}

intmain()

{

pthread_tthrd_prd_id,thrd_cst_id;

pthread_tmon_th_id;

intret;

srand(time(NULL));

end_time=time(NULL)+RUN_TIME;

/*创建有名管道*/

if((mkfifo(MYFIFO,O_CREAT|O_EXCL)0)(errno!=EEXIST))

{

printf(Cannotcreatefifon);

returnerrno;

}

/*打开管道*/

fd=open(MYFIFO,O_RDWR);

if(fd==-1)

{

printf(Openfifoerrorn);

returnfd;

}

/*初始化互斥信号量为1*/

ret=sem_init(mutex,0,1);

/*初始化avail信号量为N*/

ret+=sem_init(avail,0,BUFFER_SIZE);

/*初始化full信号量为0*/

ret+=sem_init(full,0,0);

if(ret!=0)

{

printf(Anysemaphoreinitializationfailedn);

returnret;

}

/*创建两个线程*/

ret=pthread_create(thrd_prd_id,NULL,producer,NULL);

if(ret!=0)

{

printf(Createproducerthreaderrorn);

returnret;

}

ret=pthread_create(thrd_cst_id,NULL,customer,NULL);

if(ret!=0)

{

printf(Createcustomerthreaderrorn);

returnret;

}

pthread_join(thrd_prd_id,NULL);

pthread_join(thrd_cst_id,NULL);

close(fd);

unlink(MYFIFO);

return0;

}