在Linux系統(tǒng)中玩線程，使用pthread，這篇博客記錄如何創(chuàng)建線程和使用線程和線程的同步與互斥。
還有一份nginx線程池的代碼供大家閱讀學(xué)習(xí)！
目錄
一、簡(jiǎn)介
什么是線程
線程的優(yōu)點(diǎn)、缺點(diǎn)
線程的應(yīng)用場(chǎng)合
二、線程的使用
1.? 創(chuàng)建線程 - pthread_create
2.? 線程的終止 - pthread_exit
3.? 等待指定線程結(jié)束 - pthread_join
4.? 父線程與子線程同時(shí)執(zhí)行 - pthread_detach
5.? 獲取線程id - pthread_self
6.? 比較線程id - pthread_equal
7.? 殺死線程 - pthread_cancel
8. 使用線程程序的編譯
9. 例
例1?
例2
三、線程 同步 與 互斥
1. 信號(hào)量
1). 什么是信號(hào)量
2). 信號(hào)量的初始化 - sem_init
3). 信號(hào)量的P操作 - sem_wait
4). 信號(hào)量的V操作
5). 信號(hào)量的刪除 - sem_destroy
6). 例
7). 練習(xí)
2. 互斥量
1). 什么是互斥量
2). 互斥量的初始化 - pthread_mutex_init
3). 互斥量初始化第二個(gè)參數(shù)用法?pthread_mutexattr_t *attr
4). 互斥量的獲取 - pthread_mutex_lock
5). 互斥量的釋放
6). 互斥量的刪除 - pthread_mutex_destroy
7). 例一
8). 例二
四、線程的 條件變量
1. 條件變量初始化 - pthread_cond_init
2. 喚醒一個(gè)等待線程 - pthread_cond_signal
3. 喚醒所有等待該條件變量的線程 - pthread_cond_broadcast
4. 等待條件變量 | 超時(shí)被喚醒 - pthread_cond_timedwait
5. 等待條件變量被喚醒 - pthread_cond_wait
6. 釋放/銷毀條件變量 - pthread_cond_destroy
7. 例
五、線程池
thread.h
thread_cond.c
thread_mutex.c
thread_pool.h
thread_pool.c
測(cè)試：main.c
Makefile
六、總結(jié)
一、簡(jiǎn)介
簡(jiǎn)單說(shuō)一下概念。
進(jìn)程是包含線程的，一個(gè)進(jìn)程可以有多個(gè)線程，多個(gè)線程(或一個(gè))組合成一個(gè)進(jìn)程。
線程是CPU調(diào)度和分派的基本單位，進(jìn)程是分配資源的基本單位。
一個(gè)應(yīng)用程序就是一個(gè)進(jìn)程，例如運(yùn)行一個(gè)QQ，運(yùn)行一個(gè)微信等等，都是一個(gè)進(jìn)程。
在QQ里面，我一邊接收文件，一遍與別人聊天，這就是兩個(gè)線程在同時(shí)運(yùn)行！
?
?為什么使用線程
- 使用fork創(chuàng)建進(jìn)程以執(zhí)行新的任務(wù)，該方式的代價(jià)很高；
- 多個(gè)進(jìn)程間不會(huì)直接共享內(nèi)存；
- 線程是進(jìn)程的基本執(zhí)行單元，一個(gè)進(jìn)程的所有任務(wù)都在線程中執(zhí)行，進(jìn)程要想執(zhí)行任務(wù)，必須得有線程，進(jìn)程至少要有一條線程，程序啟動(dòng)會(huì)默認(rèn)開(kāi)啟一條線程，這條線程被稱為主線程或 UI 線程。
什么是線程
????????線程，是進(jìn)程內(nèi)部的一個(gè)控制序列。
????????即使不使用線程，進(jìn)程內(nèi)部也有一個(gè)執(zhí)行線程。
注意：?jiǎn)魏颂幚砥魃?，同一個(gè)時(shí)刻只能運(yùn)行一個(gè)線程。
????????但是對(duì)于用戶而言，感覺(jué)如同同時(shí)執(zhí)行了多個(gè)線程一樣（各線程在單核CPU上切換，在一段時(shí)間內(nèi)，同時(shí)執(zhí)行了多個(gè)線程）；
線程的優(yōu)點(diǎn)、缺點(diǎn)
??? 優(yōu)點(diǎn)： 創(chuàng)建線程比創(chuàng)建進(jìn)程，開(kāi)銷要小。
??? 缺點(diǎn)：
????????1）多線程編程，需特別小心，很容易發(fā)生錯(cuò)誤；
????????2）多線程調(diào)試很困難；
????????3）把一個(gè)任務(wù)劃分為兩部分，用兩個(gè)線程在單處理器上運(yùn)行時(shí)，不一定更快；
??????????????除非能確定這兩個(gè)部分能同時(shí)執(zhí)行、且運(yùn)行在多處理器上。
線程的應(yīng)用場(chǎng)合
????????1) 需要讓用戶感覺(jué)在同時(shí)做多件事情時(shí)，?比如，處理文檔的進(jìn)程，一個(gè)線程處理用戶編輯，一個(gè)線程同時(shí)統(tǒng)計(jì)用戶的字?jǐn)?shù)；
????????2) 當(dāng)一個(gè)應(yīng)用程序，需要同時(shí)處理輸入、計(jì)算、輸出時(shí)，可開(kāi)3個(gè)線程，分別處理輸入、計(jì)算、輸出；讓用戶感覺(jué)不到等待。
????????3) 高并發(fā)編程。
為什么使用多線程
1. 避免阻塞
????????大家知道，單個(gè)進(jìn)程只有一個(gè)主線程，當(dāng)主線程阻塞的時(shí)候，整個(gè)進(jìn)程也就阻塞了，無(wú)法再去做其它的一些功能了。
2. 避免CPU空轉(zhuǎn)
????????應(yīng)用程序經(jīng)常會(huì)涉及到RPC，數(shù)據(jù)庫(kù)訪問(wèn)，磁盤IO等操作，這些操作的速度比CPU慢很多，而在等待這些響應(yīng)時(shí)，CPU卻不能去處理新的請(qǐng)求，導(dǎo)致這種單線程的應(yīng)用程序性能很差。
3. 提升效率
????????一個(gè)進(jìn)程要獨(dú)立擁有4GB的虛擬地址空間，而多個(gè)線程可以共享同一地址空間，線程的切換比進(jìn)程的切換要快得多。
二、線程的使用
1.? 創(chuàng)建線程 - pthread_create
#include <pthread.h>
int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
描述：創(chuàng)建一個(gè)新的線程；
參數(shù)
? ? ? ? thread
? ? ? ? ? ? ? ? 指向新線程的標(biāo)識(shí)符；
? ? ? ? attr
? ? ? ? ? ? ? ? 用來(lái)設(shè)置新線程的屬性；一般取默認(rèn)屬性，該參數(shù)取NULL即可；
? ? ? ? start_routine
? ? ? ? ? ? ? ? 該線程的處理函數(shù)，該函數(shù)的返回值類型和參數(shù)類型都是 void *；
? ? ? ? arg
? ? ? ? ? ? ? ? 線程處理函數(shù)，參數(shù)三(start_routine)的參數(shù)；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回非0錯(cuò)誤編號(hào)，并且 *thread 的內(nèi)容沒(méi)有定義；
注意：使用fork創(chuàng)建進(jìn)程后，進(jìn)程馬上就啟動(dòng)，是和父進(jìn)程同時(shí)執(zhí)行fork后的代碼。
? ? ? ? ? ?使用pthread_create創(chuàng)建線程后，新線程馬上就啟動(dòng)，即執(zhí)行對(duì)應(yīng)的線程處理函數(shù)。
例：
// 線程執(zhí)行函數(shù)
void *my_thread_handle(void *arg) {
// ***
}
pthread_t mythread;
int arg = 100;
int ret;
// 創(chuàng)建線程
ret = pthread_create(&mythread, 0, my_thread_handle, &arg);
if (0 != ret) {
printf("create thread failed!\n");
exit(1);
}
2.? 線程的終止 - pthread_exit
#include <pthread.h>
void pthread_exit(void *retval);
描述：在線程執(zhí)行函數(shù)內(nèi)部調(diào)用該函數(shù)，終止線程；
參數(shù)
? ? ? ? retval
? ? ? ? ? ? ? ? 線程終止后給pthread_join函數(shù)返回的一個(gè)值，不能返回局部變量；
返回值
? ? ? ? 此函數(shù)不返回給調(diào)用者；
3.? 等待指定線程結(jié)束 - pthread_join
#include <pthread.h>
int pthread_join(pthread_t thread, void **retval);
描述：父線程等待子線程結(jié)束；
參數(shù)
? ? ? ? thread
? ? ? ? ? ? ? ? 指向線程的標(biāo)識(shí)符；
? ? ? ? retval
? ? ? ? ? ? ? ? 指向該線程函數(shù)的返回值，線程函數(shù)的返回值類型為void*，所以該參數(shù)的類型為void**;
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)非0錯(cuò)誤編號(hào)；
例：
void *thread_return;
int ret = pthread_join(mythread, &thread_return);
if (0 != ret) {
printf("pthread_join failed!\n");
exit(2);
}
printf("wait thread end, return value is %d\n", *((int*)thread_return));4.? 父線程與子線程同時(shí)執(zhí)行 - pthread_detach
#include <pthread.h>
int pthread_detach(pthread_t thread);
描述：不等待子線程結(jié)束，父線程繼續(xù)往下執(zhí)行；如果父線程結(jié)束了，子線程也會(huì)結(jié)束；
參數(shù)
? ? ? ? thread
????????????????指向線程的標(biāo)識(shí)符；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)非0錯(cuò)誤編號(hào)；
例：
int ret = pthread_detach(mythread);
if (0 != ret) {
printf("pthread_detach failed!\n");
exit(2);
}
5.? 獲取線程id - pthread_self
#include <pthread.h>
pthread_t pthread_self(void);
描述：獲取調(diào)用線程的ID；在父線程調(diào)用則返回父線程id，在子線程調(diào)用則返回子線程id；
返回值
????????這個(gè)函數(shù)總是成功的，返回調(diào)用線程的ID；
6.? 比較線程id - pthread_equal
#include <pthread.h>
int pthread_equal(pthread_t t1, pthread_t t2);
描述：比較兩個(gè)線程id是否相等；
參數(shù)
? ? ? ? t1
? ? ? ? ? ? ? ? 需要比較的第一個(gè)線程id；
? ? ? ? t2
? ? ? ? ? ? ? ? 需要比較的第二個(gè)線程id；
返回值
? ? ? ? 相等：返回非0值；
? ? ? ? 不相等：返回0；
7.? 殺死線程 - pthread_cancel
#include <pthread.h>
int pthread_cancel(pthread_t thread);
描述：向線程發(fā)送取消請(qǐng)求；強(qiáng)制讓子線程停止執(zhí)行；
參數(shù)
? ? ? ? thread
????????????????指向線程的標(biāo)識(shí)符；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回一個(gè)非0錯(cuò)誤編號(hào)；
例：
int ret = pthread_cancel(mythread);
if (0 != ret) {
printf("pthread_cancel failed!\n");
exit(2);
}8. 使用線程程序的編譯
1). 編譯時(shí)，加上 -D_REENTRANT
????????功能：告訴編譯器，編譯時(shí)需要可重入功能。
? ? ? ? ? ? ? ? ? ?即使得，在編譯時(shí)，編譯部分函數(shù)的可重入版本。
注：在單線程程序中，整個(gè)程序都是順序執(zhí)行的，一個(gè)函數(shù)在同一時(shí)刻只能被一個(gè)函數(shù)調(diào)用，但在多線程中，由于并發(fā)性，一個(gè)函數(shù)可能同時(shí)被多個(gè)函數(shù)調(diào)用，此時(shí)這個(gè)函數(shù)就成了臨界資源，很容易造成調(diào)用函數(shù)處理結(jié)果的相互影響，如果一個(gè)函數(shù)在多線程并發(fā)的環(huán)境中每次被調(diào)用產(chǎn)生的結(jié)果是不確定的，我們就說(shuō)這個(gè)函數(shù)是"不可重入的"/"線程不安全"的。
2). 編譯時(shí)，指定線程庫(kù)，加上 -pthread
功能：使用系統(tǒng)默認(rèn)的NPTL線程庫(kù),
? ? ? ? ? ? 即在默認(rèn)路徑中尋找?guī)煳募ibpthread.so
? ? ? ? ? ? 默認(rèn)路徑為/usr/lib和/usr/local/lib
3). 總結(jié)，一般使用如下形式即可：
? ? ? gcc pthread2.cpp -D_REENTRANT -pthread -o a1
9. 例
例1?
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
int my_global;
void *my_thread_handle(void *arg) {
int val;
val = *((int*)arg);
printf("new thread begin, arg = %d\n", val);
my_global += val;
sleep(3);
// 線程結(jié)束
pthread_exit(&my_global);
// 這句代碼不會(huì)執(zhí)行
printf("new thread end!\n");
}
int main(void) {
pthread_t mythread;
int arg;
int ret;
void *thread_return;
arg = 100;
my_global = 1000;
printf("my_global = %d\n", my_global);
printf("ready create thread ...\n");
// 創(chuàng)建線程
ret = pthread_create(&mythread, 0, my_thread_handle, &arg);
if (0 != ret) {
printf("create thread failed!\n");
exit(1);
}
printf("wait thread finished...\n");
// 等待子線程結(jié)束
ret = pthread_join(mythread, &thread_return);
if (0 != ret) {
printf("pthread_join failed!\n");
exit(2);
}
printf("wait thread end, return value is %d\n", *((int*)thread_return));
printf("my_global = %d\n", my_global);
printf("create thread finished!\n");
return 0;
}
例2
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
void *my_thread_handle(void *arg) {
for (int i = 0; i < 10; i++) {
printf("i = %d\n", i);
sleep(1);
}
}
int main(void) {
pthread_t mythread;
int ret;
void *thread_return;
// 創(chuàng)建線程
ret = pthread_create(&mythread, 0, my_thread_handle, 0);
if (0 != ret) {
printf("create thread failed!\n");
exit(1);
}
ret = pthread_detach(mythread);
if (0 != ret) {
printf("pthread_detach failed!\n");
exit(2);
}
sleep(3);
printf("thread end.\n");
return 0;
}
因?yàn)楦妇€程結(jié)束了，所以，即使子線程還沒(méi)結(jié)束，也得強(qiáng)制結(jié)束子線程，然后結(jié)束程序！?
三、線程 同步 與 互斥
1）線程的互斥 - 指某一資源同時(shí)只允許一個(gè)訪問(wèn)者對(duì)其進(jìn)行訪問(wèn)，具有唯一性和排它性。但互斥無(wú)法限制訪問(wèn)者對(duì)資源的訪問(wèn)順序，即訪問(wèn)是無(wú)序的。
2）線程的同步 - 指在互斥的基礎(chǔ)上（大多數(shù)情況），通過(guò)其它機(jī)制實(shí)現(xiàn)訪問(wèn)者對(duì)資源的有序訪問(wèn)。
1. 問(wèn)題
??? 同一個(gè)進(jìn)程內(nèi)的各個(gè)線程，共享該進(jìn)程內(nèi)的全局變量，如果多個(gè)線程同時(shí)對(duì)某個(gè)全局變量進(jìn)行訪問(wèn)時(shí)，就可能導(dǎo)致競(jìng)態(tài)。
??? 解決辦法： 對(duì)臨界區(qū)使用信號(hào)量、或互斥量。
???
2. 信號(hào)量和互斥量的選擇。
??? 對(duì)于同步和互斥，使用信號(hào)量或互斥量都可以實(shí)現(xiàn)。
??? 使用時(shí)，選擇更符合語(yǔ)義的手段：
?????????????? 如果要求最多只允許一個(gè)線程進(jìn)入臨界區(qū)，則使用互斥量
?????????????? 如果要求多個(gè)線程之間的執(zhí)行順序滿足某個(gè)約束，則使用信號(hào)量
1. 信號(hào)量
1). 什么是信號(hào)量
???????? 此時(shí)所指的“信號(hào)量”是指用于同一個(gè)進(jìn)程內(nèi)多個(gè)線程之間的信號(hào)量。
???????? 即POSIX信號(hào)量，而不是System V信號(hào)量（用于進(jìn)程之間的同步)
????????
???????? 用于線程的信號(hào)量的原理，與用于進(jìn)程之間的信號(hào)量的原理相同。
???????? 都有P操作、V操作。
????????
???????? 信號(hào)量的表示：sem_t? 類型
2). 信號(hào)量的初始化 - sem_init
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
描述：對(duì)信號(hào)量進(jìn)行初始化；
參數(shù)
? ? ? ? sem
? ? ? ? ? ? ? ? 指向被初始化的信號(hào)量；
? ? ? ? pshared
????????????????0：表示該信號(hào)量是該進(jìn)程內(nèi)使用的“局部信號(hào)量”， 不再被其它進(jìn)程共享。
????????????????非0：該信號(hào)量可被其他進(jìn)程共享，Linux不支持這種信號(hào)量
? ? ? ? value
? ? ? ? ? ? ? ? 信號(hào)量的初始值，>= 0;
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回-1，并設(shè)置錯(cuò)誤標(biāo)志errno；
例：
sem_t sem; // 信號(hào)量標(biāo)識(shí)符，全局變量
// 初始化信號(hào)量
int ret = sem_init(&sem, 0 , 0); // 信號(hào)量值為1，第三個(gè)參數(shù)控制
if (0 != ret) {
printf("sem_init failed!\n");
exit(2);
}
3). 信號(hào)量的P操作 - sem_wait
#include <semaphore.h>
int sem_wait(sem_t *sem);
描述：執(zhí)行P操作，鎖定一個(gè)信號(hào)量；
參數(shù)
? ? ? ? sem
? ? ? ? ? ? ? ? 信號(hào)量標(biāo)識(shí)符；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回-1，并設(shè)置錯(cuò)誤標(biāo)志errno；
例：
// P操作
if (sem_wait(&sem) != 0) {
printf("sem_wait failed!\n");
exit(1);
}
4). 信號(hào)量的V操作
#include <semaphore.h>
int sem_post(sem_t *sem);
描述：執(zhí)行V操作，解鎖信號(hào)量；
參數(shù)
? ? ? ? sem
? ? ? ? ? ? ? ? 信號(hào)量標(biāo)識(shí)符；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回-1，并設(shè)置錯(cuò)誤標(biāo)志errno；
例：
// V操作
if (sem_post(&sem) != 0) {
printf("sem_post failed!\n");
exit(1);
}
5). 信號(hào)量的刪除 - sem_destroy
#include <semaphore.h>
int sem_destroy(sem_t *sem);
描述：銷毀信號(hào)量；
參數(shù)
? ? ? ? sem
? ? ? ? ? ? ? ? 信號(hào)量標(biāo)識(shí)符；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回-1，并設(shè)置錯(cuò)誤標(biāo)志errno；
例：
// 刪除信號(hào)量
int ret = sem_destroy(&sem);
if (0 != ret) {
printf("sem_destroy failed!\n");
exit(1);
}
6). 例
#include <pthread.h>
#include <semaphore.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#define BUFF_SIZE 1024
char buff[BUFF_SIZE] = { '\0' };
sem_t sem; // 信號(hào)量標(biāo)識(shí)符
static void *str_thread_handle(void *arg) {
while(1) {
// P操作
if (sem_wait(&sem) != 0) {
printf("sem_wait failed!\n");
exit(1);
}
printf("string is : %s len = %ld\n", buff, strlen(buff));
if (strncmp(buff, "end", 3) == 0) {
break;
}
}
pthread_exit(NULL);
}
int main(int argc, char **argv) {
int ret;
pthread_t str_thread;
void *thread_return;
// 初始化信號(hào)量
ret = sem_init(&sem, 0 , 0); // 信號(hào)量值為1，第三個(gè)參數(shù)控制
if (0 != ret) {
printf("sem_init failed!\n");
exit(2);
}
// 創(chuàng)建線程
ret = pthread_create(&str_thread, 0, str_thread_handle, NULL);
if (0 != ret) {
printf("pthread_create failed!\n");
exit(3);
}
while (1) {
// 輸入一行字符串
fgets(buff, sizeof(buff), stdin);
// V操作
if (sem_post(&sem) != 0) {
printf("sem_post failed!\n");
exit(4);
}
if (strncmp(buff, "end", 3) == 0) {
break;
}
}
// 父線程等待子線程
ret = pthread_join(str_thread, &thread_return);
if(0 != ret) {
printf("pthread_join failed!\n");
exit(5);
}
// 刪除信號(hào)量
ret = sem_destroy(&sem);
if (0 != ret) {
printf("sem_destroy failed!\n");
exit(6);
}
return 0;
}
7). 練習(xí)
創(chuàng)建2個(gè)線程（共有主線程、線程1、線程2共3個(gè)線程）；
主線程阻塞式等待用戶輸入字符串；
主線程每接收到一個(gè)字符串之后， 線程1就馬上對(duì)該字符串進(jìn)行處理；
線程1的處理邏輯為：統(tǒng)計(jì)該字符串的個(gè)數(shù)，并記錄當(dāng)時(shí)的時(shí)間；
線程1把該字符串處理完后，線程2馬上就把處理結(jié)果寫入文件result.txt；
直到用戶輸入exit.
#include <pthread.h>
#include <semaphore.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#define BUFF_SIZE 1024
char buff[BUFF_SIZE] = { '\0' };
sem_t sem; // 信號(hào)量標(biāo)識(shí)符
int count = 0;
char time_[BUFF_SIZE] = { '\0' };
static void *str_thread_handle1(void *arg) {
while(1) {
// P操作
if (sem_wait(&sem) != 0) {
printf("sem_wait failed!\n");
exit(1);
}
count = strlen(buff);
time_t timep;
time (&timep);
sprintf(time_, "%s", asctime( gmtime(&timep) ));
// V操作
if (sem_post(&sem) != 0) {
printf("sem_post failed!\n");
exit(4);
}
if (strncmp(buff, "exit", 4) == 0) {
break;
}
}
pthread_exit(NULL);
}
static void *str_thread_handle2(void *arg) {
FILE *fp = 0; //定義文件指針fp
// 以只讀的方式打開(kāi)文件
if ( (fp = fopen("./result.txt","a")) == 0 ) {
printf("文件打開(kāi)失敗\n");
return (void*)-1;
}
while(1) {
// P操作
if (sem_wait(&sem) != 0) {
printf("sem_wait failed!\n");
exit(1);
}
char str[BUFF_SIZE] = { '\0' };
sprintf(str, "內(nèi)容：%s大?。?d \t 時(shí)間：%s\n", buff, count, time_);
if (strncmp(buff, "exit", 4) == 0) {
break;
}
// 寫入文件
fwrite(str, strlen(str), 1, fp);
}
fclose(fp);
pthread_exit(NULL);
}
int main(int argc, char **argv) {
int ret;
pthread_t str_thread1;
pthread_t str_thread2;
void *thread_return;
// 初始化信號(hào)量
ret = sem_init(&sem, 0 , 0); // 信號(hào)量值為1，第三個(gè)參數(shù)控制
if (0 != ret) {
printf("sem_init failed!\n");
exit(2);
}
// 創(chuàng)建線程1
ret = pthread_create(&str_thread1, 0, str_thread_handle1, NULL);
if (0 != ret) {
printf("pthread_create failed!\n");
exit(3);
}
// 創(chuàng)建線程2
ret = pthread_create(&str_thread2, 0, str_thread_handle2, NULL);
if (0 != ret) {
printf("pthread_create failed!\n");
exit(3);
}
while (1) {
// 輸入一行字符串
fgets(buff, sizeof(buff), stdin);
// V操作
if (sem_post(&sem) != 0) {
printf("sem_post failed!\n");
exit(4);
}
if (strncmp(buff, "exit", 4) == 0) {
break;
}
}
// 父線程等待子線程
ret = pthread_join(str_thread1, &thread_return);
if(0 != ret) {
printf("pthread_join failed!\n");
exit(5);
}
ret = pthread_join(str_thread2, &thread_return);
if(0 != ret) {
printf("pthread_join failed!\n");
exit(5);
}
// 刪除信號(hào)量
ret = sem_destroy(&sem);
if (0 != ret) {
printf("sem_destroy failed!\n");
exit(6);
}
return 0;
}
2. 互斥量
1). 什么是互斥量
???????? 效果上等同于初值為1的信號(hào)量
???????? 互斥量的使用：類型為 pthread_mutex_t
2). 互斥量的初始化 - pthread_mutex_init
#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);
描述：初始化互斥對(duì)象；
參數(shù)
? ? ? ? mutex
? ? ? ? ? ? ? ? 指向被初始化的互斥量；
? ? ? ? attr
? ? ? ? ? ? ? ? 指向互斥量的屬性，一般取默認(rèn)值（當(dāng)一個(gè)線程已獲取互斥量后，該線程再次獲取該信號(hào)量，將導(dǎo)致死鎖!);
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回非0錯(cuò)誤碼；
例：
pthread_mutex_t lock; // 定義互斥鎖，全局變量
// 初始化互斥鎖
int ret = pthread_mutex_init(&lock, 0);
if (0 != ret) {
printf("pthread_mutex_ini failed!\n");
exit(1);
}
3). 互斥量初始化第二個(gè)參數(shù)用法?pthread_mutexattr_t *attr
將attr值設(shè)置為：PTHREAD_MUTEX_ERRORCHECK
即如果使用給互斥鎖上鎖兩次，第二次上鎖時(shí)就會(huì)報(bào)錯(cuò)！
例：可以先看完下面的用法之后再返回來(lái)看這里的代碼
#include <pthread.h>
#include <semaphore.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
pthread_mutex_t lock; // 定義互斥鎖
static void *str_thread_handle(void *arg) {
// do something...
// 退出子線程
pthread_exit(NULL);
}
int main(int argc, char **argv) {
int ret;
pthread_t tid; // the thread identifier
void *thread_return;
pthread_mutexattr_t attr;
// 初始化屬性
ret = pthread_mutexattr_init(&attr);
if (ret != 0) {
fprintf(stderr, "pthread_mutexattr_init() failed, reason: %s\n",strerror(errno));
exit(1);
}
// 設(shè)置如果給互斥鎖上鎖兩次，就會(huì)報(bào)錯(cuò)！
ret = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
if (ret != 0) {
fprintf(stderr, "pthread_mutexattr_settype(PTHREAD_MUTEX_ERRORCHECK) failed, reason: %s\n",strerror(errno));
exit(1);
}
// 初始化互斥鎖
ret = pthread_mutex_init(&lock, &attr);
if (0 != ret) {
printf("pthread_mutex_ini failed!\n");
exit(1);
} else {
// 銷毀屬性
err = pthread_mutexattr_destroy(&attr);
if (err != 0) {
fprintf(stderr,"pthread_mutexattr_destroy() failed, reason: %s\n",strerror(errno));
}
}
// 創(chuàng)建線程
ret = pthread_create(&tid, 0, str_thread_handle, 0);
if (0 != ret) {
printf("pthread_create failed!\n");
exit(2);
}
// 第一次上鎖
ret = pthread_mutex_lock(&lock);
if (0 != ret) {
printf("pthread_mutex_lock failed!\n");
exit(1);
}
printf("-----------------------------------start two lock.\n");
// 沒(méi)有解鎖，進(jìn)行第二次上鎖
ret = pthread_mutex_lock(&lock); // 這里會(huì)返回錯(cuò)誤
if (0 != ret) {
printf("pthread_mutex_lock failed!\n");
exit(1);
}
// 等待子線程
ret = pthread_join(tid, &thread_return);
if (0 != ret) {
printf("pthread_join failed!\n");
exit(3);
}
// 銷毀互斥鎖
ret = pthread_mutex_destroy(&lock);
if (0 != ret) {
printf("pthread_mutex_destroy failed!\n");
exit(4);
}
return 0;
}4). 互斥量的獲取 - pthread_mutex_lock
#include <pthread.h>
int pthread_mutex_lock(pthread_mutex_t *mutex);
描述：鎖定互斥量；
參數(shù)
? ? ? ? mutex
? ? ? ? ? ? ? ? 定義的全局互斥量；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回非0錯(cuò)誤碼；
例：
// 上鎖
int ret = pthread_mutex_lock(&lock);
if (0 != ret) {
printf("pthread_mutex_lock failed!\n");
exit(1);
}
5). 互斥量的釋放
#include <pthread.h>
int pthread_mutex_unlock(pthread_mutex_t *mutex);
描述：解鎖互斥鎖；
參數(shù)
? ? ? ? mutex
? ? ? ? ? ? ? ? 定義的全局互斥量；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回非0錯(cuò)誤碼；
例：
// 解鎖
int ret = pthread_mutex_unlock(&lock);
if (0 != ret) {
printf("pthread_mutex_unlock failed!\n");
exit(1);
}
6). 互斥量的刪除 - pthread_mutex_destroy
#include <pthread.h>
int pthread_mutex_destroy(pthread_mutex_t *mutex);
描述：銷毀互斥對(duì)象；
參數(shù)
? ? ? ? mutex
? ? ? ? ? ? ? ? 定義的全局互斥量；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回非0錯(cuò)誤碼；
例：
// 銷毀互斥鎖
int ret = pthread_mutex_destroy(&lock);
if (0 != ret) {
printf("pthread_mutex_destroy failed!\n");
exit(1);
}
7). 例一
#include <pthread.h>
#include <semaphore.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
int global_value = 0;
pthread_mutex_t lock; // 定義互斥鎖
static void *str_thread_handle(void *arg) {
int i = 0;
for (i = 0; i < 10; i++) {
// 上鎖
pthread_mutex_lock(&lock);
sleep(1);
global_value++;
printf("【Child pthread】 ：global_value = %d\n", global_value);
// 解鎖
pthread_mutex_unlock(&lock);
sleep(1);
}
// 退出子線程
pthread_exit(NULL);
}
int main(int argc, char **argv) {
int ret;
pthread_t tid; // the thread identifier
void *thread_return;
int i;
// 初始化互斥鎖
ret = pthread_mutex_init(&lock, 0);
if (0 != ret) {
printf("pthread_mutex_ini failed!\n");
exit(1);
}
ret = pthread_create(&tid, 0, str_thread_handle, 0);
if (0 != ret) {
printf("pthread_create failed!\n");
exit(2);
}
for (i = 0; i < 10; i++) {
// 上鎖
ret = pthread_mutex_lock(&lock);
if (0 != ret) {
printf("pthread_mutex_lock failed!\n");
exit(1);
}
sleep(1);
global_value++;
printf("【Father pthread】 ：global_value = %d\n", global_value);
// 解鎖
ret = pthread_mutex_unlock(&lock);
if (0 != ret) {
printf("pthread_mutex_unlock failed!\n");
exit(1);
}
sleep(1);
}
// 等待子線程
ret = pthread_join(tid, &thread_return);
if (0 != ret) {
printf("pthread_join failed!\n");
exit(3);
}
// 銷毀互斥鎖
ret = pthread_mutex_destroy(&lock);
if (0 != ret) {
printf("pthread_mutex_destroy failed!\n");
exit(4);
}
return 0;
}
8). 例二
主線程輸入字符串，子線程統(tǒng)計(jì)字符串個(gè)數(shù)
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
#include <errno.h>
void *thread_function(void *arg);
pthread_mutex_t work_mutex; // 定義互斥鎖
#define WORK_SIZE 1024
char work_area[WORK_SIZE] = { '\0' };
int main(int argc, char **argv) {
int res;
pthread_t a_thread;
void *thread_result;
// 初始化互斥鎖
res = pthread_mutex_init(&work_mutex, NULL);
if (0 != res) {
perror("Mutex initialization failed!\n");
exit(EXIT_FAILURE);
}
// 創(chuàng)建線程
res = pthread_create(&a_thread, NULL, thread_function, NULL);
if (0 != res) {
perror("Thread creation failed!\n");
exit(EXIT_FAILURE);
}
printf("Input some text. Enter 'end' to finish!\n");
while (1) {
// 上鎖
pthread_mutex_lock(&work_mutex);
fgets(work_area, WORK_SIZE, stdin); // 輸入一行字符串
if (strncmp("end", work_area, 3) == 0) {
break;
}
// 解鎖
pthread_mutex_unlock(&work_mutex);
sleep(1);
}
// break 后需要解鎖
pthread_mutex_unlock(&work_mutex);
printf("\nWaiting for thread to finish...\n");
// 等待子線程
res = pthread_join(a_thread, &thread_result);
if (0 != res) {
perror("Thread join failed!\n");
exit(EXIT_FAILURE);
}
printf("Thread joined!\n");
// 銷毀互斥鎖
res = pthread_mutex_destroy(&work_mutex);
if (0 != res) {
perror("pthread_mutex_destroy failed!\n");
exit(EXIT_FAILURE);
}
exit(EXIT_SUCCESS);
}
void *thread_function(void *arg) {
sleep(1);
while (1) {
// 上鎖
pthread_mutex_lock(&work_mutex);
if (strncmp("end", work_area, 3) == 0) {
break;
}
printf("Uou input %ld characters\n", strlen(work_area) - 1);
// 解鎖
pthread_mutex_unlock(&work_mutex);
sleep(1);
}
// break 后需要解鎖
pthread_mutex_unlock(&work_mutex);
pthread_exit(0);
}
四、線程的 條件變量
與互斥鎖不同，條件變量是用來(lái)等待而不是用來(lái)上鎖的。條件變量用來(lái)自動(dòng)阻塞一個(gè)線程，直到某特殊情況發(fā)生為止。通常條件變量和互斥鎖同時(shí)使用。
條件變量使我們可以睡眠等待某種條件出現(xiàn)。條件變量是利用線程間共享的全局變量進(jìn)行同步的一種機(jī)制，主要包括兩個(gè)動(dòng)作：一個(gè)線程等待"條件變量的條件成立"而掛起；另一個(gè)線程使"條件成立"（給出條件成立信號(hào)）。
條件的檢測(cè)是在互斥鎖的保護(hù)下進(jìn)行的。如果一個(gè)條件為假，一個(gè)線程自動(dòng)阻塞，并釋放等待狀態(tài)改變的互斥鎖。如果另一個(gè)線程改變了條件，它發(fā)信號(hào)給關(guān)聯(lián)的條件變量，喚醒一個(gè)或多個(gè)等待它的線程，重新獲得互斥鎖，重新評(píng)價(jià)條件。如果兩進(jìn)程共享可讀寫的內(nèi)存，條件變量可以被用來(lái)實(shí)現(xiàn)這兩進(jìn)程間的線程同步。
安裝文檔手冊(cè)命令：apt-get install manpages-posix-dev
1. 條件變量初始化 - pthread_cond_init
#include <pthread.h>
int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr);
描述：初始化條件變量；
參數(shù)
? ? ? ? cond
? ? ? ? ? ? ? ? 條件變量指針；
? ? ? ? attr
? ? ? ? ? ? ? ? 條件變量高級(jí)屬性，設(shè)置為NULL即可；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)錯(cuò)誤數(shù)字；
例：
pthread_cond_t cond; // 條件變量，全局
int ret = pthread_cond_init(&cond, NULL);
if (0 != ret) {
printf("pthread_cond_init failed!\n");
exit(1);
}2. 喚醒一個(gè)等待線程 - pthread_cond_signal
#include <pthread.h>
int pthread_cond_signal(pthread_cond_t *cond);
描述：發(fā)出條件信號(hào)；通知條件變量，喚醒一個(gè)等待的線程；
參數(shù)
????????cond
? ? ? ? ? ? ? ? 條件變量指針；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)錯(cuò)誤數(shù)字；
例：
int ret = pthread_cond_signal(&cond);
if (0 != ret) {
printf("pthread_cond_signal failed!\n");
exit(1);
}3. 喚醒所有等待該條件變量的線程 - pthread_cond_broadcast
#include <pthread.h>
int pthread_cond_broadcast(pthread_cond_t *cond);
描述：廣播條件變量；即所有等待的線程都可以收到；
參數(shù)
????????cond
? ? ? ? ? ? ? ? 條件變量指針；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)錯(cuò)誤數(shù)字；
例：
int ret = pthread_cond_broadcast(&cond);
if (0 != ret) {
printf("pthread_cond_broadcast failed!\n");
exit(1);
}4. 等待條件變量 | 超時(shí)被喚醒 - pthread_cond_timedwait
#include <pthread.h>
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime);
描述：等待條件變量被喚醒；或者絕對(duì)時(shí)間abstime到了喚醒該線程；
參數(shù)
????????cond
? ? ? ? ? ? ? ? 條件變量指針；
? ? ? ? mutex
? ? ? ? ? ? ? ? 互斥量指針；
? ? ? ? abstime
? ? ? ? ? ? ? ? 等待被喚醒的絕對(duì)超時(shí)時(shí)間；
????????????????struct?timespec {
????????????????????????time_t tv_sec; // seconds
????????????????????????long tv_nsec; // and nanoseconds
????????????????};
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)錯(cuò)誤數(shù)字；
5. 等待條件變量被喚醒 - pthread_cond_wait
#include <pthread.h>
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
描述：等待條件變量cond被喚醒(由一個(gè)信號(hào)或這廣播喚醒);
參數(shù)
????????cond
? ? ? ? ? ? ? ? 條件變量指針；
? ? ? ? mutex
? ? ? ? ? ? ? ? 互斥量指針；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失敗：返回一個(gè)錯(cuò)誤數(shù)字；
例：
pthread_mutex_t mutex; // 互斥鎖，全局變量
pthread_cond_t cond; // 條件變量，全局變量
int ret = pthread_cond_wait(&cond, &mutex);
if (0 != ret) {
printf("pthread_cond_wait failed!\n");
exit(1);
}為什么需要傳入互斥量指針呢？
pthread_cond_wait 執(zhí)行過(guò)程：
1). 掛起當(dāng)前線程；
2). 使用互斥量進(jìn)行解鎖；
3). 當(dāng)有信號(hào)通知后；
4). 喚醒線程；
5). 使用互斥量進(jìn)行上鎖；
6). 才會(huì)往下執(zhí)行其他代碼；
由此可知，有多個(gè)線程執(zhí)行等待時(shí)，是可以同時(shí)等待的，但如果只有一個(gè)信號(hào)過(guò)來(lái)，也就只能有一個(gè)線程被喚醒；注意，即使使用pthread_cond_broadcast 廣播發(fā)送信號(hào)去喚醒全部線程，線程的執(zhí)行也只能一個(gè)一個(gè)的去執(zhí)行，因?yàn)橛谢コ饬康木壒剩?/p>
6. 釋放/銷毀條件變量 - pthread_cond_destroy
#include <pthread.h>
int pthread_cond_destroy(pthread_cond_t *cond);
描述：銷毀條件變量；
參數(shù)
? ? ? ? cond
? ? ? ? ? ? ? ? 條件變量指針；
返回值
? ? ? ? 成功：返回0；
? ? ? ? 失?。?/span>返回一個(gè)錯(cuò)誤數(shù)字；
例：
int ret = pthread_cond_destroy(&cond);
if (0 != ret) {
printf("pthread_cond_destroy failed!\n");
exit(1);
}
7. 例
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
pthread_mutex_t mutex; // 互斥鎖
pthread_cond_t cond; // 條件變量
void *thread1(void *arg) {
while (1) {
printf("thread1 is running\n");
// 上鎖
pthread_mutex_lock(&mutex);
// 條件變量，等待信號(hào)觸發(fā)往下執(zhí)行
pthread_cond_wait(&cond, &mutex);
printf("thread1 applied the condition\n\n");
// 解鎖
pthread_mutex_unlock(&mutex);
sleep(4);
}
}
void *thread2(void *arg) {
while (1) {
printf("thread2 is running\n");
// 上鎖
pthread_mutex_lock(&mutex);
// 條件變量，等待信號(hào)觸發(fā)往下執(zhí)行
pthread_cond_wait(&cond, &mutex);
printf("thread2 applied the condition\n\n");
// 解鎖
pthread_mutex_unlock(&mutex);
sleep(2);
}
}
int main(int argc, char **argv) {
pthread_t thid1, thid2; // 線程變量
printf("condition variable study!\n");
// 互斥鎖初始化
pthread_mutex_init(&mutex, NULL);
// 條件變量初始化
pthread_cond_init(&cond, NULL);
// 創(chuàng)建線程
pthread_create(&thid1, NULL, thread1, NULL);
pthread_create(&thid2, NULL, thread2, NULL);
int i = 0;
do {
pthread_cond_signal(&cond);
sleep(1);
} while(i++ < 10);
int ret = pthread_cond_destroy(&cond);
if (0 != ret) {
printf("pthread_cond_destroy failed!\n");
exit(1);
}
return 0;
}
五、線程池
以下代碼時(shí)NginX的線程池代碼；
寫的還是挺不錯(cuò)的，B格挺高的！
有興趣的可以閱讀一下！
另外，可以直接復(fù)制到自己的項(xiàng)目中去使用！
thread.h
#ifndef _DEMO_THREAD_H_INCLUDED_
#define _DEMO_THREAD_H_INCLUDED_
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <sys/types.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>
typedef intptr_t int_t;
typedef uintptr_t uint_t;
#define OK 0
#define ERROR -1
int thread_mutex_create(pthread_mutex_t *mtx);
int thread_mutex_destroy(pthread_mutex_t *mtx);
int thread_mutex_lock(pthread_mutex_t *mtx);
int thread_mutex_unlock(pthread_mutex_t *mtx);
int thread_cond_create(pthread_cond_t *cond);
int thread_cond_destroy(pthread_cond_t *cond);
int thread_cond_signal(pthread_cond_t *cond);
int thread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mtx);
#endif /* _DEMO_THREAD_H_INCLUDED_ */thread_cond.c
#include "thread.h"
int
thread_cond_create(pthread_cond_t *cond)
{
int err;
err = pthread_cond_init(cond, NULL);
if (err == 0) {
return OK;
}
fprintf(stderr, "pthread_cond_init() failed, reason: %s\n",strerror(errno));
return ERROR;
}
int
thread_cond_destroy(pthread_cond_t *cond)
{
int err;
err = pthread_cond_destroy(cond);
if (err == 0) {
return OK;
}
fprintf(stderr, "pthread_cond_destroy() failed, reason: %s\n",strerror(errno));
return ERROR;
}
int
thread_cond_signal(pthread_cond_t *cond)
{
int err;
err = pthread_cond_signal(cond);
if (err == 0) {
return OK;
}
fprintf(stderr, "pthread_cond_signal() failed, reason: %s\n",strerror(errno));
return ERROR;
}
int
thread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mtx)
{
int err;
err = pthread_cond_wait(cond, mtx);
if (err == 0) {
return OK;
}
fprintf(stderr, "pthread_cond_wait() failed, reason: %s\n",strerror(errno));
return ERROR;
}
thread_mutex.c
#include "thread.h"
int
thread_mutex_create(pthread_mutex_t *mtx)
{
int err;
pthread_mutexattr_t attr;
err = pthread_mutexattr_init(&attr);
if (err != 0) {
fprintf(stderr, "pthread_mutexattr_init() failed, reason: %s\n",strerror(errno));
return ERROR;
}
err = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
if (err != 0) {
fprintf(stderr, "pthread_mutexattr_settype(PTHREAD_MUTEX_ERRORCHECK) failed, reason: %s\n",strerror(errno));
return ERROR;
}
err = pthread_mutex_init(mtx, &attr);
if (err != 0) {
fprintf(stderr,"pthread_mutex_init() failed, reason: %s\n",strerror(errno));
return ERROR;
}
err = pthread_mutexattr_destroy(&attr);
if (err != 0) {
fprintf(stderr,"pthread_mutexattr_destroy() failed, reason: %s\n",strerror(errno));
}
return OK;
}
int
thread_mutex_destroy(pthread_mutex_t *mtx)
{
int err;
err = pthread_mutex_destroy(mtx);
if (err != 0) {
fprintf(stderr,"pthread_mutex_destroy() failed, reason: %s\n",strerror(errno));
return ERROR;
}
return OK;
}
int
thread_mutex_lock(pthread_mutex_t *mtx)
{
int err;
err = pthread_mutex_lock(mtx);
if (err == 0) {
return OK;
}
fprintf(stderr,"pthread_mutex_lock() failed, reason: %s\n",strerror(errno));
return ERROR;
}
int
thread_mutex_unlock(pthread_mutex_t *mtx)
{
int err;
err = pthread_mutex_unlock(mtx);
#if 0
ngx_time_update();
#endif
if (err == 0) {
return OK;
}
fprintf(stderr,"pthread_mutex_unlock() failed, reason: %s\n",strerror(errno));
return ERROR;
}
thread_pool.h
#ifndef _THREAD_POOL_H_INCLUDED_
#define _THREAD_POOL_H_INCLUDED_
#include "thread.h"
#define DEFAULT_THREADS_NUM 4 // 線程池中線程個(gè)數(shù)
#define DEFAULT_QUEUE_NUM 65535 // 任務(wù)隊(duì)列最大值
typedef unsigned long atomic_uint_t;
typedef struct thread_task_s thread_task_t; // 任務(wù)
typedef struct thread_pool_s thread_pool_t; // 線程池
struct thread_task_s {
thread_task_t *next;
uint_t id;
void *ctx; // 參數(shù)地址的位置
void (*handler)(void *data);
};
typedef struct {
thread_task_t *first;
thread_task_t **last;
} thread_pool_queue_t;
#define thread_pool_queue_init(q) \
(q)->first = NULL; \
(q)->last = &(q)->first
struct thread_pool_s {
pthread_mutex_t mtx; // 互斥量，與條件變量一起使用
thread_pool_queue_t queue; // 任務(wù)隊(duì)列
int_t waiting; // 任務(wù)隊(duì)列等待的個(gè)數(shù)
pthread_cond_t cond; // 條件變量，與互斥量一同使用
char *name; // 名字，可不指定
uint_t threads; // 線程池中的線程個(gè)數(shù)
int_t max_queue; // 任務(wù)隊(duì)列最大值
};
thread_task_t *thread_task_alloc(size_t size);
void thread_task_free(thread_task_t *task);
int_t thread_task_post(thread_pool_t *tp, thread_task_t *task);
thread_pool_t* thread_pool_init();
void thread_pool_destroy(thread_pool_t *tp);
#endif /* _THREAD_POOL_H_INCLUDED_ */
thread_pool.c
#include "thread_pool.h"
static void thread_pool_exit_handler(void *data);
static void *thread_pool_cycle(void *data);
static int_t thread_pool_init_default(thread_pool_t *tpp, char *name);
static uint_t thread_pool_task_id;
static int debug = 0;
thread_pool_t* thread_pool_init()
{
int err;
pthread_t tid;
uint_t n;
pthread_attr_t attr;
thread_pool_t *tp=NULL;
tp = calloc(1,sizeof(thread_pool_t));
if(tp == NULL){
fprintf(stderr, "thread_pool_init: calloc failed!\n");
}
thread_pool_init_default(tp, NULL);
thread_pool_queue_init(&tp->queue);
// 初始化互斥量
if (thread_mutex_create(&tp->mtx) != OK) {
free(tp);
return NULL;
}
// 初始化條件變量
if (thread_cond_create(&tp->cond) != OK) {
(void) thread_mutex_destroy(&tp->mtx);
free(tp);
return NULL;
}
err = pthread_attr_init(&attr);
if (err) {
fprintf(stderr, "pthread_attr_init() failed, reason: %s\n",strerror(errno));
free(tp);
return NULL;
}
// PTHREAD_CREATE_DETACHED：創(chuàng)建的子線程與主線程分離，即主線程不等待子線程結(jié)束，各自運(yùn)行各自的
err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
if (err) {
fprintf(stderr, "pthread_attr_setdetachstate() failed, reason: %s\n",strerror(errno));
free(tp);
return NULL;
}
for (n = 0; n < tp->threads; n++) {
err = pthread_create(&tid, &attr, thread_pool_cycle, tp);
if (err) {
fprintf(stderr, "pthread_create() failed, reason: %s\n",strerror(errno));
free(tp);
return NULL;
}
}
(void) pthread_attr_destroy(&attr);
return tp;
}
void thread_pool_destroy(thread_pool_t *tp)
{
uint_t n;
thread_task_t task;
volatile uint_t lock;
memset(&task,'\0', sizeof(thread_task_t));
task.handler = thread_pool_exit_handler;
task.ctx = (void *) &lock;
for (n = 0; n < tp->threads; n++) {
lock = 1;
if (thread_task_post(tp, &task) != OK) {
return;
}
while (lock) {
sched_yield();
}
//task.event.active = 0;
}
(void) thread_cond_destroy(&tp->cond);
(void) thread_mutex_destroy(&tp->mtx);
free(tp);
}
static void
thread_pool_exit_handler(void *data)
{
uint_t *lock = data;
*lock = 0;
pthread_exit(0);
}
thread_task_t *
thread_task_alloc(size_t size)
{
thread_task_t *task;
// 分配的內(nèi)存 加上 線程運(yùn)行時(shí)傳入的參數(shù)的位置大小
task = calloc(1,sizeof(thread_task_t) + size);
if (task == NULL) {
return NULL;
}
task->ctx = task + 1; // task + 1：就指向了參數(shù)的起始位置
return task;
}
void thread_task_free(thread_task_t *task) {
if (task) {
free(task);
task = NULL;
}
}
int_t
thread_task_post(thread_pool_t *tp, thread_task_t *task)
{
if (thread_mutex_lock(&tp->mtx) != OK) {
return ERROR;
}
if (tp->waiting >= tp->max_queue) {
(void) thread_mutex_unlock(&tp->mtx);
fprintf(stderr,"thread pool \"%s\" queue overflow: %ld tasks waiting\n",
tp->name, tp->waiting);
return ERROR;
}
//task->event.active = 1;
task->id = thread_pool_task_id++;
task->next = NULL;
if (thread_cond_signal(&tp->cond) != OK) {
(void) thread_mutex_unlock(&tp->mtx);
return ERROR;
}
*tp->queue.last = task; // 鏈接到尾部
tp->queue.last = &task->next; // 指向尾部
tp->waiting++;
(void) thread_mutex_unlock(&tp->mtx);
if(debug)fprintf(stderr,"task #%lu added to thread pool \"%s\"\n",
task->id, tp->name);
return OK;
}
static void *
thread_pool_cycle(void *data)
{
thread_pool_t *tp = data;
int err;
thread_task_t *task;
if(debug)fprintf(stderr,"thread in pool \"%s\" started\n", tp->name);
for ( ;; ) {
if (thread_mutex_lock(&tp->mtx) != OK) {
return NULL;
}
tp->waiting--;
while (tp->queue.first == NULL) {
if (thread_cond_wait(&tp->cond, &tp->mtx)
!= OK)
{
(void) thread_mutex_unlock(&tp->mtx);
return NULL;
}
}
task = tp->queue.first;
tp->queue.first = task->next;
if (tp->queue.first == NULL) {
tp->queue.last = &tp->queue.first;
}
if (thread_mutex_unlock(&tp->mtx) != OK) {
return NULL;
}
if(debug) fprintf(stderr,"run task #%lu in thread pool \"%s\"\n",
task->id, tp->name);
task->handler(task->ctx);
if(debug) fprintf(stderr,"complete task #%lu in thread pool \"%s\"\n",task->id, tp->name);
task->next = NULL;
thread_task_free(task);
//notify
}
}
static int_t
thread_pool_init_default(thread_pool_t *tpp, char *name)
{
if(tpp)
{
tpp->threads = DEFAULT_THREADS_NUM;
tpp->max_queue = DEFAULT_QUEUE_NUM;
tpp->name = strdup(name?name:"default");
if(debug)fprintf(stderr,
"thread_pool_init, name: %s ,threads: %lu max_queue: %ld\n",
tpp->name, tpp->threads, tpp->max_queue);
return OK;
}
return ERROR;
}
測(cè)試：main.c
#include "thread_pool.h"
#include <unistd.h>
// 線程參數(shù)
struct test{
int arg1;
int arg2;
};
void task_handler1(void* data){
static int index = 0;
printf("Hello, this is 1th test.index=%d\r\n", index++);
}
void task_handler2(void* data){
static int index = 0;
printf("Hello, this is 2th test.index=%d\r\n", index++);
}
void task_handler3(void* data){
static int index = 0;
struct test *t = (struct test *) data;
printf("Hello, this is 3th test.index=%d\r\n", index++);
printf("arg1: %d, arg2: %d\n", t->arg1, t->arg2);
}
int
main(int argc, char **argv)
{
thread_pool_t* tp = NULL;
int i = 0;
tp = thread_pool_init();
//sleep(1);
thread_task_t * test1 = thread_task_alloc(0);
thread_task_t * test2 = thread_task_alloc(0);
thread_task_t * test3 = thread_task_alloc(sizeof(struct test));
test1->handler = task_handler1;
test2->handler = task_handler2;
test3->handler = task_handler3;
((struct test*)test3->ctx)->arg1 = 666;
((struct test*)test3->ctx)->arg2 = 888;
//for(i=0; i<10;i++){
thread_task_post(tp, test1);
thread_task_post(tp, test2);
thread_task_post(tp, test3);
//}
sleep(5);
thread_pool_destroy(tp);
}Makefile
CXX := gcc
TARGET = threadpool
RMRF := rm -rf
SRC = $(wildcard *.cpp *.c) # 獲取當(dāng)前路徑下的所有.cpp 和 .c 文件
OBJ = $(patsubst %.cpp %.c, %.o, $(SRC)) # 把$(SRC)中符合.cpp 和 .c 的所有文件轉(zhuǎn)換成.o文件
CXXFLAGS = -c -Wall
LIBS = -lpthread
$(TARGET): $(OBJ)
$(CXX) $^ $(LIBS) -o $@
%.o: %cpp %c
$(CXX) $(CXXFLAGS) $< -o $@
.PHONY: clean
clean:
$(RMRF) *.o
cleanall:
$(RMRF) *.o $(TARGET)
六、總結(jié)
線程的操作已經(jīng)完畢；個(gè)人覺(jué)得互斥量是線程經(jīng)常使用到的，所以得更加專注的去學(xué)習(xí)那一塊知識(shí)點(diǎn)！文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-431311.html
另外，線程池這部分代碼，效率還是挺高的，所以也得去學(xué)習(xí)一下大佬的寫法！文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-431311.html
到了這里，關(guān)于Linux pthread線程操作 和 線程同步與互斥操作的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
