pthread多線程(C語言) + Socket

pthread是使用使用C語言編寫的多線程的API， 簡稱Pthreads ，是線程的POSIX標(biāo)準(zhǔn)，可以在Unix / Linux / Windows 等系統(tǒng)跨平臺(tái)使用。在類Unix操作系統(tǒng)（Unix、Linux、Mac OS X等）中，都使用Pthreads作為操作系統(tǒng)的線程。

GitHub項(xiàng)目FanSocket(純C語言socket+線程隊(duì)列)+其他demo客戶端

1.線程創(chuàng)建

//子線程1
void test1(int *a){
    printf("線程test1");
    //修改自己的子線程系統(tǒng)釋放，注釋打開后，線程不能用pthread_join方法
    //pthread_detach(pthread_self());
}
//子線程2
void test2(int *a){
    printf("線程test2");
}

/*
int  pthread_create(pthread_t  *  thread, //新線程標(biāo)識(shí)符
pthread_attr_t * attr, //新線程的運(yùn)行屬性
void * (*start_routine)(void *), //線程將會(huì)執(zhí)行的函數(shù)
void * arg);//執(zhí)行函數(shù)的傳入?yún)?shù)，可以為結(jié)構(gòu)體
*/

//創(chuàng)建線程方法一    （手動(dòng)釋放線程）
int a=10;
pthread_t pid;
pthread_create(&pid, NULL, (void *)test1, (void *)&a);


//線程退出或返回時(shí)，才執(zhí)行回調(diào)，可以釋放線程占用的堆棧資源（有串行的作用）
if(pthread_join(pid, NULL)==0){
    //線程執(zhí)行完成
    printf("線程執(zhí)行完成：%d\n",threadIndex);
    if (message!=NULL) {
        printf("線程執(zhí)行完成了\n");
    }
}


//創(chuàng)建線程方法二  （自動(dòng)釋放線程）
//設(shè)置線程屬性
pthread_attr_t attr;
pthread_attr_init (&attr);
//線程默認(rèn)是PTHREAD_CREATE_JOINABLE，需要pthread_join來釋放線程的
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
//線程并發(fā)
int rc=pthread_create(&pid, &attr, (void *)test2, (void *)a);
pthread_attr_destroy (&attr);
if (rc!=0) {
    printf("創(chuàng)建線程失敗\n");
    return;
}

2.線程退出和其他

pthread_exit (tes1) //退出當(dāng)前線程
pthread_main_np () // 獲取主線程

//主線程和子線程
if(pthread_main_np()){
    //main thread
}else{
    //others thread
}

int pthread_cancel(pthread_t thread)；//發(fā)送終止信號(hào)給thread線程,如果成功則返回0
int pthread_setcancelstate(int state, int *oldstate)；//設(shè)置本線程對(duì)Cancel信號(hào)的反應(yīng)
int pthread_setcanceltype(int type, int *oldtype)；//設(shè)置本線程取消動(dòng)作的執(zhí)行時(shí)機(jī)
void pthread_testcancel(void)；//檢查本線程是否處于Canceld狀態(tài)，如果是，則進(jìn)行取消動(dòng)作，否則直接返回

3 線程互斥鎖（量）與條件變量

3.1 互斥鎖(量)

//靜態(tài)創(chuàng)建
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
//動(dòng)態(tài)創(chuàng)建
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *mutexattr);
//注銷互斥鎖
int pthread_mutex_destroy(pthread_mutex_t *mutex);

//lock 和unlock要成對(duì)出現(xiàn)，不然會(huì)出現(xiàn)死鎖
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
//判斷是否可以加鎖，如果可以加鎖并返回0，否則返回非0
int pthread_mutex_trylock(pthread_mutex_t *mutex);

3.2 條件變量

• 條件變量是利用線程間共享的全局變量進(jìn)行同步的一種機(jī)制，
• 一個(gè)線程等待”條件變量的條件成立”而掛起；
• 另一個(gè)線程使”條件成立”（給出條件成立信號(hào)）。
• 為了防止競(jìng)爭(zhēng)，條件變量的使用總是和一個(gè)互斥鎖結(jié)合在一起。

//靜態(tài)創(chuàng)建
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;
//動(dòng)態(tài)創(chuàng)建
int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);
//注銷條件變量
int pthread_cond_destroy(pthread_cond_t *cond);
//條件等待，和超時(shí)等待
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, 
                           const struct timespec *abstime);
//開啟條件，啟動(dòng)所有等待線程
int pthread_cond_broadcast(pthread_cond_t *cond);
//開啟一個(gè)等待信號(hào)量
int pthread_cond_signal(pthread_cond_t *cond);

4.線程同步：互斥鎖（量）與條件變量(具體封裝實(shí)現(xiàn))

/*全局的隊(duì)列互斥條件*/
extern pthread_cond_t fan_cond;
extern pthread_cond_t fan_cond_wait;
/*全局的隊(duì)列互斥鎖*/
extern pthread_mutex_t fan_mutex;
//extern pthread_mutex_t fan_mutex_wait;

extern int fan_thread_status;//0=等待 1=執(zhí)行 -1=清空所有
extern int fan_thread_clean_status;//0=默認(rèn)  1=清空所有

//開啟線程等待  return=-2一定要處理
extern int fan_thread_start_wait(void);
//正常的超時(shí)后繼續(xù)打開下一個(gè)信號(hào)量 return=-2一定要處理
int fan_thread_start_timedwait(int sec);
//啟動(dòng)線程，啟動(dòng)信號(hào)量
extern int fan_thread_start_signal(void);
//啟動(dòng)等待信號(hào)量
extern int fan_thread_start_signal_wait(void);
//暫停線程
extern int fan_thread_end_signal(void);
//初始化互斥鎖
extern int fan_thread_queue_init(void);
//釋放互斥鎖信號(hào)量
extern int fan_thread_free(void);

//讓隊(duì)列里面全部執(zhí)行完畢，而不是關(guān)閉線程；
extern int fan_thread_clean_queue(void);
//每次關(guān)閉清空后，等待1-2秒，要恢復(fù)狀態(tài)，不然線程添加
extern int fan_thread_init_queue(void);
//設(shè)置線程的優(yōu)先級(jí)，必須在子線程
extern int fan_thread_setpriority(int priority);

線程隊(duì)列互斥，并且按入隊(duì)順序，一個(gè)一個(gè)按照外部條件，觸發(fā)信號(hào)量，主要是等待隊(duì)列，

/*全局的隊(duì)列互斥條件*/
pthread_cond_t fan_cond=PTHREAD_COND_INITIALIZER;
pthread_cond_t fan_cond_wait=PTHREAD_COND_INITIALIZER;

/*全局的隊(duì)列互斥鎖*/
pthread_mutex_t fan_mutex = PTHREAD_MUTEX_INITIALIZER;
//pthread_mutex_t fan_mutex_wait = PTHREAD_MUTEX_INITIALIZER;
int fan_thread_status=1;//0=等待 1=執(zhí)行
int fan_thread_clean_status;//0=默認(rèn)  1=清空所有

//開啟線程等待
int fan_thread_start_wait(void){
    pthread_mutex_lock(&fan_mutex);
    fan_thread_clean_status=0;
    while (fan_thread_status==0) {
        pthread_cond_wait(&fan_cond, &fan_mutex);
        if (fan_thread_clean_status==1) {
            break;
        }
    }
    if (fan_thread_clean_status==1) {
        pthread_mutex_unlock(&fan_mutex);
        return -2;
    }
    if (fan_thread_status==1) {
        fan_thread_status=0;
        pthread_mutex_unlock(&fan_mutex);
    }else{
        pthread_mutex_unlock(&fan_mutex);
    }
    return 0;
}
//正常的超時(shí)后繼續(xù)打開下一個(gè)信號(hào)量
int fan_thread_start_timedwait(int sec){
    int rt=0;
    pthread_mutex_lock(&fan_mutex);
    struct timeval now;
    struct timespec outtime;
    gettimeofday(&now, NULL);
    outtime.tv_sec = now.tv_sec + sec;
    outtime.tv_nsec = now.tv_usec * 1000;
    
    int result = pthread_cond_timedwait(&fan_cond_wait, &fan_mutex, &outtime);
    if (result!=0) {
        //線程等待超時(shí)
        rt=-1;
    }
    if (fan_thread_clean_status==1) {
        rt = -2;
    }
    pthread_mutex_unlock(&fan_mutex);
    return rt;
}
//啟動(dòng)線程，啟動(dòng)信號(hào)量
int fan_thread_start_signal(void){
    int rs=pthread_mutex_trylock(&fan_mutex);
    if(rs!=0){
        pthread_mutex_unlock(&fan_mutex);
    }
    fan_thread_status=1;
    pthread_cond_signal(&fan_cond);
//    pthread_cond_broadcast(&fan_cond);//全部線程
    pthread_mutex_unlock(&fan_mutex);
    return 0;
}
//開啟等待時(shí)間的互斥信號(hào)量
int fan_thread_start_signal_wait(void){
    int rs=pthread_mutex_trylock(&fan_mutex);
    if(rs!=0){
        pthread_mutex_unlock(&fan_mutex);
    }
//    fan_thread_status=1;
    pthread_cond_signal(&fan_cond_wait);
    //    pthread_cond_broadcast(&fan_cond);//全部線程
    pthread_mutex_unlock(&fan_mutex);
    return 0;
}
//暫停下一個(gè)線程
int fan_thread_end_signal(void){
    pthread_mutex_lock(&fan_mutex);
    fan_thread_status=0;
    pthread_cond_signal(&fan_cond);
    pthread_mutex_unlock(&fan_mutex);
    return 0;
}
//初始化互斥鎖（動(dòng)態(tài)創(chuàng)建）
int fan_thread_queue_init(void){
    pthread_mutex_init(&fan_mutex, NULL);
    pthread_cond_init(&fan_cond, NULL);
    return 0;
}
//釋放互斥鎖和信號(hào)量
int fan_thread_free(void)
{
    pthread_mutex_destroy(&fan_mutex);
    pthread_cond_destroy(&fan_cond);
    return 0;
}

//清空所有的隊(duì)列
int fan_thread_clean_queue(void){
    pthread_mutex_lock(&fan_mutex);
    fan_thread_clean_status=1;
    pthread_cond_broadcast(&fan_cond);
    pthread_cond_broadcast(&fan_cond_wait);
    pthread_mutex_unlock(&fan_mutex);
    return 0;
}
//恢復(fù)隊(duì)列
int fan_thread_init_queue(void){
    pthread_mutex_lock(&fan_mutex);
    fan_thread_clean_status=0;
    fan_thread_status=1;
    pthread_cond_signal(&fan_cond);
    pthread_mutex_unlock(&fan_mutex);
    return 0;
}
//設(shè)置線程的優(yōu)先級(jí)，必須在子線程
int fan_thread_setpriority(int priority){
    struct sched_param sched;
    bzero((void*)&sched, sizeof(sched));
//    const int priority1 = (sched_get_priority_max(SCHED_RR) + sched_get_priority_min(SCHED_RR)) / 2;
    sched.sched_priority=priority;
    //SCHED_OTHER(正常,非實(shí)時(shí))SCHED_FIFO(實(shí)時(shí)，先進(jìn)先出)SCHED_RR(實(shí)時(shí)、輪轉(zhuǎn)法)
    pthread_setschedparam(pthread_self(), SCHED_RR, &sched);
    return 0;
}

5 其他線程方法

//return=0:線程存活。ESRCH：線程不存在。EINVAL：信號(hào)不合法。
int kill_ret=pthread_kill(pid, 0);//測(cè)試線程是否存在
printf("線程狀態(tài)：%d\n",kill_ret);
if(kill_ret==0){
    //關(guān)閉線程
    pthread_cancel(pid);
}


pthread_equal(pid, pid1);//比較兩個(gè)線程ID是否相同


//函數(shù)執(zhí)行一次
pthread_once_t once = PTHREAD_ONCE_INIT;
pthread_once(&once, test1);