多線程編程之：實(shí)驗(yàn)內(nèi)容——“生產(chǎn)者消費(fèi)者”實(shí)驗(yàn)

（3）編寫代碼

本實(shí)驗(yàn)的代碼中采用的有界緩沖區(qū)擁有3個(gè)單元，每個(gè)單元為5個(gè)字節(jié)。為了盡量體現(xiàn)每個(gè)信號量的意義，在程序中生產(chǎn)過程和消費(fèi)過程是隨機(jī)（采取0～5s的隨機(jī)時(shí)間間隔）進(jìn)行的，而且生產(chǎn)者的速度比消費(fèi)者的速度平均快兩倍左右（這種關(guān)系可以相反）。生產(chǎn)者一次生產(chǎn)一個(gè)單元的產(chǎn)品（放入“hello”字符串），消費(fèi)者一次消費(fèi)一個(gè)單元的產(chǎn)品。

/*producer-customer.c*/

#includestdio.h>

#includestdlib.h>

#includeunistd.h>

#includefcntl.h>

#includepthread.h>

#includeerrno.h>

#includesemaphore.h>

#includesys/ipc.h>

#defineMYFIFOmyfifo/*緩沖區(qū)有名管道的名字*/

#defineBUFFER_SIZE3/*緩沖區(qū)的單元數(shù)*/

#defineUNIT_SIZE5/*每個(gè)單元的大小*/

#defineRUN_TIME30/*運(yùn)行時(shí)間*/

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

intfd;

time_tend_time;

sem_tmutex,full,avail;/*3個(gè)信號量*/

/*生產(chǎn)者線程*/

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);

/*生產(chǎn)者寫入數(shù)據(jù)*/

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);

}

/*消費(fèi)者線程*/

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;

/*創(chuàng)建有名管道*/

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;

}

/*創(chuàng)建兩個(gè)線程*/

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;

}