linux多线程学习（七）——实现“生产者和消费者”

在上一篇文章中，利用信号量实现了线程间的互斥，这一篇将要利用信号量的互斥同步机制来实现一个经典实例，就是“生产者和消费者”。

1、简单描述生产者和消费者的问题。

有一个缓冲区和两个线程：生产者和消费者。生产者把产品放入缓冲区，而消费者从缓冲区中拿走。当缓冲区满时，生产者必须等待；另外，当缓冲区空时，消费者必须等待，并且缓冲区不能同时进行生产者和消费者的操作。

[cpp:nogutter] view
plaincopy

1. #include <stdio.h>    
2. #include <stdlib.h>    
3. #include <pthread.h>    
4. #include <semaphore.h>    
5. #include <errno.h>    
6. #include <fcntl.h>    
7. #include <string.h>  
8.     
9. #define FIFO "my_fifo"    
10. #define N    10    
11. #define return_if_fail(p) if((p) == 0){printf ("[%s]:func error!/n", __func__);return;}    
12.     
13. typedef struct _PrivInfo    
14. {    
15.   sem_t avail;    
16.   sem_t full;    
17.   sem_t mutex;  
18.   char  buf_r[255];    
19.   time_t end_time;    
20.   int fd;    
21. }PrivInfo;    
22.     
23. static void info_init (PrivInfo* thiz);    
24. static void info_destroy (PrivInfo* thiz);    
25. static void* productor (PrivInfo* thiz);    
26. static void* consumer (PrivInfo* thiz);    
27.     
28. int main (int argc, char* argv[])    
29. {    
30.   pthread_t pt_1 = 0;    
31.   pthread_t pt_2 = 0;    
32.   int ret = 0;    
33.   PrivInfo* thiz = NULL;    
34.       
35.   thiz = (PrivInfo*) malloc (sizeof (PrivInfo));    
36.   if (thiz == NULL)    
37.   {    
38.     printf ("[%s]: Failed to malloc PrivInfo./n", __func__);    
39.     return -1;    
40.   }    
41.     
42.   info_init (thiz);    
43.     
44.   ret = pthread_create (&pt_1, NULL, (void*)productor, thiz);    
45.   if (ret != 0)    
46.   {    
47.     perror ("pthread_1_create:");    
48.   }    
49.     
50.   ret = pthread_create (&pt_2, NULL, (void*)consumer, thiz);    
51.   if (ret != 0)    
52.   {    
53.     perror ("pthread_2_func:");    
54.   }    
55.     
56.   pthread_join (pt_1, NULL);    
57.   pthread_join (pt_2, NULL);    
58.     
59.   info_destroy (thiz);    
60.   thiz = NULL;    
61.     
62.   return 0;    
63. }    
64.     
65. static void info_init (PrivInfo* thiz)    
66. {    
67.   return_if_fail (thiz != NULL);    
68.     
69.   thiz->end_time = time(NULL) + 10;    
70.       
71.   sem_init (&thiz->avail, 0, N);    
72.   sem_init (&thiz->full, 0, 0);    
73.   sem_init (&thiz->mutex, 0, 1);    
74.     
75.   if (mkfifo (FIFO, O_CREAT|O_EXCL) && (errno != EEXIST))    
76.   {    
77.     printf ("cannot create fifo server./n");    
78.   }    
79.   
80.   printf ("Prepareing for reading bytes.../n");    
81.     
82.   memset (thiz->buf_r, 0, sizeof (thiz->buf_r));    
83.     
84.   thiz->fd = open (FIFO, O_RDWR|O_NONBLOCK, 0777);    
85.   if (thiz->fd == -1)    
86.   {    
87.     perror ("open FIFO:");    
88.     info_destroy (thiz);    
89.     exit (1);    
90.   }    
91.     
92.   return ;    
93. }    
94.     
95. static void info_destroy (PrivInfo* thiz)    
96. {    
97.   return_if_fail (thiz != NULL);    
98.     
99.   sem_destroy (&thiz->avail);    
100.   sem_destroy (&thiz->full);    
101.   sem_destroy (&thiz->mutex);    
102.     
103.   free (thiz);    
104.   thiz = NULL;    
105.     
106.   return;    
107. }    
108.     
109. static void* productor (PrivInfo* thiz)    
110. {    
111.   return_if_fail (thiz != NULL);    
112.     
113.  int ret = 0;    
114.     
115.   while (time (NULL) < thiz->end_time)    
116.   {    
117.     sem_wait (&thiz->avail);    
118.     sem_wait (&thiz->mutex);    
119.     
120.     if ((ret = write (thiz->fd, "hello", 5)) == -1)    
121.     {    
122.        if (errno == EAGAIN)    
123.        {    
124.            printf ("The FIFO has not been read, Please try later again./n");    
125.        }    
126.        else    
127.        {    
128.            printf ("write hello to the FIFO./n");    
129.        }    
130.     }    
131.     
132.     sem_post (&thiz->full);    
133.     sem_post (&thiz->mutex);    
134.     
135.     sleep (1);    
136.   }    
137.     
138.   return;    
139. }    
140.     
141. static void* consumer (PrivInfo* thiz)    
142. {    
143.    return_if_fail (thiz != NULL);    
144.     
145.    int ret = 0;    
146.     
147.    while (time (NULL) < thiz->end_time)    
148.   {    
149.      sem_wait (&thiz->full);    
150.      sem_wait (&thiz->mutex);    
151.     
152.      if ((ret = read(thiz->fd, thiz->buf_r, 255)) > 0)    
153.      {    
154.                   
155.         printf ("read %s frm FIFO./n", thiz->buf_r);    
156.       }    
157.       else   
158.       {  
159.         if (errno == EAGAIN)    
160.         {    
161.            printf ("no data yet./n");     
162.         }  
163.       }  
164.     
165.       sem_post (&thiz->avail);    
166.       sem_post (&thiz->mutex);    
167.   
168.       sleep(1);  
169.   }    
170.     
171.   return;    
172. }    

从上面的经典例子中，结合了多线程的同时性编程问题。学习线程算是告了一段落，在以后的编程应用中，希望能落实到实践中去，大家一起加油~~