1 #include <stdio.h>
2 #include <unistd.h>
3 #include <stdlib.h>
4 #include <pthread.h>
5
6 typedef void* (*fun)(void*);
7
8 fun fun1, fun2;
9
10 pthread_mutex_t pmu = PTHREAD_MUTEX_INITIALIZER;
11 pthread_cond_t cond;
12 pthread_t pid1, pid2;
13 int flag = 0;
14 int gnum = 0;
15 int gsub = 100;
16
17 void * func1(void * para)
18 {
19 int k = (int)para;
20 printf("func1, ******\n");
21 while(gnum<=100)
22 {
23 pthread_mutex_lock(&pmu);
24 printf("gnum == %d", gnum);
25 while(gnum==50)
26 {
27 printf("suspend thread1 at gnum==50 !!! \n");
28 pthread_cond_wait(&cond, &pmu);
29 gnum++;
30 }
31 ++gnum;
32 ++flag;
33 ++k;
34 //printf("flag = %d, k = %d\n", flag, k);
35 pthread_mutex_unlock(&pmu);
36 printf("I am func1\n");
37 }
38 pthread_exit((void*)0);
39
40 }
41
42 void * func2(void * para)
43 {
44 int f = (int)para;
45 printf("f == %d\n", f);
46 printf("pthread2 start running !\n");
47 void * ret = NULL;
48 while(gsub>=0)
49 {
50 pthread_mutex_lock(&pmu);
51 gsub--;
52 printf("gsub= %d ", gsub);
53 if(gsub == 20)
54 {
55 printf("now gsnb ==20, and send signal\n");
56 pthread_cond_signal(&cond);
57 }
58 ++flag;
59 ++f;
60 printf("flag = %d, f = %d\n", flag, f);
61 pthread_mutex_unlock(&pmu);
62 printf("I am func2 \n");
63 }
64 //pthread_join(pid1, &ret);
65 pthread_exit((void*)0);
66 }
67
68 int main()
69 {
70 int id = 0;
71 void * ret = NULL;
72 int key = 5;
73
74 pthread_cond_init(&cond, NULL); //属性设置NULL默认属性
75 id = pthread_create(&pid1, NULL, func1, (void*)key);
76 if(id != 0)
77 {
78 printf("pthread_create error !\n");
79 exit(0);
80 }
81
82 if(pthread_create(&pid2, NULL, func2, (void*)key))
83 {
84 printf("pthread_create error ! \n");
85 exit(0);
86 }
87 pthread_join(pid2, &ret); //等待pid2线程退出
88 pthread_join(pid1, &ret); //等待pid1线程退出
//pthread_detach(pid1); //主线程与pid1线程进行分离，一般用来实现异步返回 //pthread_detach(pid2); //同上
89 pthread_exit((void*)0);
90 }
gcc test_thread.c -lpthread
./a.out
线程池实例代码：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <sys/types.h>
5 #include <pthread.h>
6 #include <assert.h>
7
8 typedef struct worker
9 {
10 //回调函数，任务运行时会调用此函数，也可以声明为其他形式；
11 void * (*process)(void *arg); //该函数返回值是任意类型的；参数也是任意类型的；`
12 void *arg; //回调函数的参数；
13 struct worker *next;
14 }CThread_worker;
15
16 //线程池结构
17 typedef struct
18 {
19 pthread_mutex_t queue_lock; //互斥量
20 pthread_cond_t queue_ready; //条件变量
21
22 //链表结构， 线程池中所有等待任务
23 CThread_worker *queue_head;
24
25 //是否销毁线程池
26 int shutdown;
27 pthread_t *threadid;
28
29 //线程池中允许的活动线程数目；
30 //线程池中允许的活动线程数目；
31 int max_thread_num;
32 //当前等待队列的任务数目;
33 int cur_queue_size;
34
35 }CThread_pool;
36
37 int pool_add_worker(void * (*process)(void *arg), void *arg);
38 void * thread_routine(void *arg);
39
40 static CThread_pool *pool = NULL;
41 void pool_init(int max_thread_num)
42 {
43 pool = (CThread_pool*)malloc(sizeof(CThread_pool));
44
45 //初始化互斥量；
46 pthread_mutex_init(&(pool->queue_lock), NULL);
47 //初始化条件变量
48 pthread_cond_init(&(pool->queue_ready), NULL);
49
50 pool->queue_head = NULL;
51
52 //最大线程数目
53 pool->max_thread_num = max_thread_num;
54 //当前线程数目
55 pool->cur_queue_size = 0;
56
57 pool->shutdown = 0;
58 pool->threadid = (pthread_t*)malloc(max_thread_num * sizeof(pthread_t));
59 int i = 0;
60 for(i=0; i<max_thread_num;i++)
61 {
62 pthread_create(&(pool->threadid[i]), NULL, thread_routine, NULL);
63 }
64 }
65
66 //向线程池中加入任务
67 int pool_add_worker(void*(*process)(void *arg), void *arg)
68 {
69 //构建一个新任务
70 CThread_worker *newworker = (CThread_worker *)malloc(sizeof(CThread_worker));
71 newworker->process = process;
72 newworker->arg = arg;
73 //别忘了置空
74 newworker->next = NULL;
75
76 //加锁互斥量
77 pthread_mutex_lock(&(pool->queue_lock));
78 //将任务加入到等待队列中
79 CThread_worker *member = pool->queue_head;
80 if(member !=NULL)
81 {
82 while(member->next != NULL)
83 member = member->next;
84 member->next = newworker;
85 }
86 else
87 {
88 pool->queue_head = newworker;
89 }
90
91 assert(pool->queue_head != NULL);
92 pool->cur_queue_size++;
93 pthread_mutex_unlock(&(pool->queue_lock));
94
95 //好了，等待队列中有任务了，唤醒一个等待线程；
96 // 注意如果所有线程都在忙碌，这句没有任何作用
97 pthread_cond_signal(&(pool->queue_ready));
98 return 0;
99 }
100
101 /*销毁线程池，等待队列中的任务不会再被执行，
102 *但是正在运行的线程会一直 把任务运行完后 再退出；
103 */
104
105 int pool_destroy()
106 {
107 if(pool->shutdown)
108 return -1; //防止两次调用
109 pool->shutdown = 1;
110
111 //唤醒所有等待线程，线程池要销毁了
112 pthread_cond_broadcast(&(pool->queue_ready));
113
114 //阻塞等待线程退出， 否则就成僵尸了
115 int i;
116 for(i=0; i<pool->max_thread_num; i++)
117 {
118 pthread_join(pool->threadid[i], NULL);
119 }
120
121 free(pool->threadid);
122
123 //销毁等待队列
124 CThread_worker *head = NULL;
125 while(pool->queue_head != NULL)
126 {
127 head=pool->queue_head;
128 pool->queue_head = pool->queue_head->next;
129 free(head);
130 }
131
132 //条件变量和互斥量也别忘了销毁
133 pthread_mutex_destroy(&(pool->queue_lock));
134 pthread_cond_destroy(&(pool->queue_ready));
135
136 free(pool);
137 /*销毁后指针置空是个好习惯*/
138 pool = NULL;
139 return 0;
140 }
141
142 void* thread_routine(void *arg)
143 {
144 printf("start thread 0x%x\n", pthread_self());
145 while(1)
146 {
147 pthread_mutex_lock(&(pool->queue_lock));
148 /*如果等待队列为0并且不销毁线程池，则处于阻塞状态; 注意
149 *pthread_cond_wait是一个原子操作，等待前会解锁，唤醒后会加锁*/
150 while(pool->cur_queue_size == 0 && !pool->shutdown)
151 {
152 printf("thread 0x%x is waiting \n", pthread_self());
153 pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
154 }
155
156 //线程池要销毁了；
157 if(pool->shutdown)
158 {
159 //遇到break,continue,return等跳转语句，千万不要忘记先解锁*/
160 pthread_mutex_unlock(&(pool->queue_lock));
161 printf("thread 0x %x will exit \n", pthread_self());
162 pthread_exit(NULL);
163 }
164
165 printf("thread 0x %x is starting to work \n", pthread_self());
166
167 //使用断言
168 assert(pool->cur_queue_size!= 0);
169 assert(pool->queue_head!= NULL);
170
171 //等待队列长度减去1，并取出链表中的头元素
172 pool->cur_queue_size--;
173 CThread_worker *worker = pool->queue_head;
174 pool->queue_head = worker->next;
175 pthread_mutex_unlock(&(pool->queue_lock));
176
177 //调用回调函数，执行任务
178 (*(worker->process))(worker->arg);
179 free(worker);
180 worker = NULL;
181 }
182 //这一句正常情况下是不可达的
183 pthread_exit(NULL);
184 }
185
186 //test code
187 void *myprocess(void *arg)
188 {
189 printf("threadid is 0x%x, working on task %d\n", pthread_self(), *(int*)arg);
190 sleep(1); //休息一秒，延长任务的执行时间
191 return NULL;
192 }
193
194 int main(int argc, char** argv)
195 {
196 pool_init(3); /*线程池中最多三个活动线程*/
197
198 //连续向线程池中放入10个任务；
199 int *workingnum = (int*)malloc(sizeof(int)*10);
200 int i;
201 for(i=0; i< 10;i++)
202 {
203 workingnum[i] = i;
204 pool_add_worker(myprocess, &workingnum[i]);
205 }
206
207 sleep(5);
208 //销毁线程池；
209 pool_destroy();
210 free(workingnum);
211
212 return 0;
213 }
时间： 2025-01-01 06:48:06