網(wǎng)絡(luò)技術(shù)編程之多線程服務(wù)器端的實(shí)現(xiàn)

2015-10-28 08:48 作者：admin 瀏覽量：

　　線程的創(chuàng)建和上下文切換開銷更小且速度更快。（網(wǎng)絡(luò)外包公司）

　　線程間交換數(shù)據(jù)時(shí)無需特殊技術(shù)。

　　進(jìn)程：在操作系統(tǒng)構(gòu)成單獨(dú)執(zhí)行流的單位。

　　線程：在進(jìn)程構(gòu)成單獨(dú)執(zhí)行流的單位。

　　它們的包含關(guān)系是，操作系統(tǒng) > 進(jìn)程 > 線程。進(jìn)程與線程具體差異其實(shí)是這樣的，每個(gè)進(jìn)程都有獨(dú)立的完整內(nèi)存空間，它包括全局?jǐn)?shù)據(jù)區(qū)，堆區(qū)，棧區(qū)，而多進(jìn)程服務(wù)器之所以開銷大是因?yàn)橹皇菫榱藚^(qū)分棧區(qū)里的不同函數(shù)流執(zhí)行而把數(shù)據(jù)區(qū)，堆區(qū)，棧區(qū)內(nèi)存全部復(fù)制了一份。而多線程就高效多了，它只把棧區(qū)分離出來，進(jìn)程中的數(shù)據(jù)區(qū)，堆區(qū)則共享。具體內(nèi)存結(jié)構(gòu)示例圖如下：

　　線程創(chuàng)建及運(yùn)行

　　線程具有單獨(dú)的執(zhí)行流，因此需要單獨(dú)定義線程的入口函數(shù)，而且還需要請(qǐng)求操作系統(tǒng)在單獨(dú)的執(zhí)行流中執(zhí)行該函數(shù)，完成這個(gè)功能的函數(shù)如下：

　　#include int pthread_create( pthread_t * restrict thread,//保存線程ID const pthread_attr_t * restrict attr,//線程屬性，NULL默認(rèn)屬性 void * (* start_routine)(void *), //線程入口函數(shù)，函數(shù)指針 void * restrict arg //傳遞給入口函數(shù)的參數(shù) );

　　實(shí)例代碼：

　　#include #include #include void * thread_main(void *arg); int main(int argc, char *argv[]) { pthread_t t_id; int thread_param = 5; if (pthread_create(&t_id, NULL, thread_main, (void *)&thread_param) != 0) { puts("pthread_create() error"); return -1; } sleep(10); puts("end of main"); return 0; } void * thread_main(void *arg) { int i; int cnt =* ((int *)arg); for (i = 0; i < cnt; i++) { sleep(1); puts("running thread"); } return NULL; }

　　上面實(shí)例是用sleep延遲來控制線程的執(zhí)行的，如果主線程不做延遲那么執(zhí)行到return 0;時(shí)，進(jìn)程就結(jié)束了，相應(yīng)的線程也會(huì)銷毀。而明顯用sleep這種方式控制線程執(zhí)行流是不合理的，下面我們來看看一個(gè)更好的延遲函數(shù)，調(diào)用該函數(shù)的進(jìn)程(或線程)將進(jìn)入等待狀態(tài)，直到第一個(gè)參數(shù)為ID的線程終止為止。而且可以得到線程的入口函數(shù)返回值。（IT外包）

　　#include int pthread_join(pthread_t thread, void **status); 參數(shù)1：線程ID 參數(shù)2：保存線程入口函數(shù)的返回值

　　實(shí)例代碼：

　　#include #include #include #include void * thread_main(void *arg); int main(int argc, char *argv[]) { pthread_t t_id; int thread_param = 5; void * thr_ret; //創(chuàng)建線程 if (pthread_create(&t_id, NULL, thread_main, (void *)&thread_param) != 0) { puts("pthread_create() error"); return -1; } //等待線程返回 if (pthread_join(t_id, &thr_ret) != 0) { puts("pthread_join() error"); return -1; } printf("Thread return message: %s \n", (char *)thr_ret); free(thr_ret); return 0; } //線程入口函數(shù) void * thread_main(void *arg) { int i; int cnt =* ((int *)arg); char * msg = (char *)malloc(sizeof(char) * 50); strcpy(msg, "Hello, I am thread ~ \n"); for (i = 0; i < cnt; i++) { puts("running thread"); } return (void *)msg; }

　　線程存在的問題和臨界區(qū)

　　前面我們知道了怎么創(chuàng)建線程，但我們都是只創(chuàng)建了一個(gè)線程，下面我們?cè)賮砜纯催@樣一個(gè)實(shí)例，創(chuàng)建100個(gè)線程，它們都訪問了同一變量，其中一半對(duì)這個(gè)變量進(jìn)行加1操作，一半進(jìn)行減1操作，按道理其結(jié)果會(huì)等于0.

　　#include #include #include #include #define NUM_THREAD 100 void * thread_inc(void * arg); void * thread_des(void * arg); long long num = 0; //long long類型是64位整數(shù)型，多線程共同訪問 int main(int argc, char *argv[]) { pthread_t thread_id[NUM_THREAD]; int i; //創(chuàng)建100個(gè)線程，一半執(zhí)行thread_inc,一半執(zhí)行thread_des for(i = 0; i < NUM_THREAD; i++) { if(i %2) pthread_create(&(thread_id[i]), NULL, thread_inc, NULL); else pthread_create(&(thread_id[i]), NULL, thread_des, NULL); } //等待線程返回 for (i = 0; i < NUM_THREAD; i++) pthread_join(thread_id[i], NULL); printf("result: %lld \n", num); //+1，-1按道理結(jié)果是0 return 0; } //線程入口函數(shù)1 void * thread_inc(void * arg) { for (int i = 0; i < 50000000; i++) num += 1;//臨界區(qū)(引起問題的語句就是臨界區(qū)位置) return NULL; } //線程入口函數(shù)2 void * thread_des(void * arg) { for (int i = 0; i < 50000000; i++) num -= 1;//臨界區(qū) return NULL; }

　　從運(yùn)行結(jié)果看并不是0，而且每次運(yùn)行的結(jié)果都不同。那這是什么原因引起的呢? 是因?yàn)槊總€(gè)線程訪問一個(gè)變量是這樣一個(gè)過程：先從內(nèi)存取出這個(gè)變量值到CPU，然后CPU計(jì)算得到改變后的值，最后再將這個(gè)改變后的值寫回內(nèi)存。因此，我們可以很容易看出，多個(gè)線程訪問同一變量，如果某個(gè)線程還只剛從內(nèi)存取出數(shù)據(jù)，還沒來得及寫回內(nèi)存，這時(shí)其它線程又訪問了這個(gè)變量，所以這個(gè)值就會(huì)不正確了。

　　接下來我們?cè)賮碇v講怎么解決這個(gè)問題：線程同步

　　線程同步

　　線程同步用于解決線程訪問順序引發(fā)的問題，一般是如下兩種情況：

　　同時(shí)訪問同一內(nèi)存空間時(shí)發(fā)生的情況

　　需要指定訪問同一內(nèi)存空間的線程執(zhí)行順序的情況

　　針對(duì)這兩種可能引發(fā)的情況，我們分別使用的同步技術(shù)是：互斥量和信號(hào)量。

　　互斥量

　　互斥量技術(shù)從字面也可以理解，就是臨界區(qū)有線程訪問，其它線程就得排隊(duì)等待，它們的訪問是互斥的，實(shí)現(xiàn)方式就是給臨界區(qū)加鎖與釋放鎖。

　　#include int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr); //創(chuàng)建互斥量 int pthread_mutex_destroy(pthread_mutex_t *mutex);//銷毀互斥量 int pthread_mutex_lock(pthread_mutex_t *mutex);//加鎖 int pthread_mutex_unlock(pthread_mutex_t *mutex);//釋放鎖

　　簡(jiǎn)言之，就是利用lock和unlock函數(shù)圍住臨界區(qū)的兩端。當(dāng)某個(gè)線程調(diào)用pthread_mutex_lock進(jìn)入臨界區(qū)后，如果沒有調(diào)用pthread_mutex_unlock釋放鎖退出，那么其它線程就會(huì)一直阻塞在臨界區(qū)之外，我們把這種情況稱之為死鎖。所以臨界區(qū)圍住一定要lock和unlock一一對(duì)應(yīng)。

　　實(shí)例代碼：

　　#include #include #include #include #define NUM_THREAD 100 void * thread_inc(void * arg); void * thread_des(void * arg); long long num = 0; pthread_mutex_t mutex; int main(int argc, char *argv[]) { pthread_t thread_id[NUM_THREAD]; int i; //互斥量的創(chuàng)建 pthread_mutex_init(&mutex, NULL); for(i = 0; i < NUM_THREAD; i++) { if(i %2) pthread_create(&(thread_id[i]), NULL, thread_inc, NULL); else pthread_create(&(thread_id[i]), NULL, thread_des, NULL); } for (i = 0; i < NUM_THREAD; i++) pthread_join(thread_id[i], NULL); printf("result: %lld \n", num); pthread_mutex_destroy(&mutex); //互斥量的銷毀 return 0; } /*擴(kuò)展臨界區(qū)，減少加鎖，釋放鎖調(diào)用次數(shù)，但這樣變量必須加滿到50000000次后其它線程才能訪問. 這樣是延長(zhǎng)了線程的等待時(shí)間，但縮短了加鎖，釋放鎖函數(shù)調(diào)用的時(shí)間，這里沒有定論，自己酌情考慮*/ void * thread_inc(void * arg) { pthread_mutex_lock(&mutex); //互斥量鎖住 for (int i = 0; i < 1000000; i++) num += 1; pthread_mutex_unlock(&mutex); //互斥量釋放鎖 return NULL; } /*縮短了線程等待時(shí)間，但循環(huán)創(chuàng)建，釋放鎖函數(shù)調(diào)用時(shí)間增加*/ void * thread_des(void * arg) { for (int i = 0; i < 1000000; i++) { pthread_mutex_lock(&mutex); num -= 1; pthread_mutex_unlock(&mutex); } return NULL; }

　　信號(hào)量

　　信號(hào)量與互斥量類似，只是互斥量是用鎖來控制線程訪問而信號(hào)量是用二進(jìn)制0，1來完成控制線程順序。sem_post信號(hào)量加1，sem_wait信號(hào)量減1，當(dāng)信號(hào)量為0時(shí)，sem_wait就會(huì)阻斷，因此通過這樣讓信號(hào)量加1減1就能控制線程的執(zhí)行順序了。

　　注釋：mac上測(cè)試信號(hào)量函數(shù)返回-1失敗，以后還是Linux上整吧，也許這些接口已經(jīng)過時(shí)了…

　　#include int sem_init(sem_t *sem, int pshared, unsigned int value);//創(chuàng)建信號(hào)量 int sem_destroy(sem_t *sem);//銷毀信號(hào)量 int sem_post(sem_t *sem);//信號(hào)量加1 int sem_wait(sem_t *sem);//信號(hào)量減1，為0時(shí)阻塞

　　實(shí)例代碼：線程A從用戶輸入得到值后存入全局變量num，此時(shí)線程B將取走該值并累加。該過程共進(jìn)行5次，完成后輸出總和并退出程序。

　　#include #include #include void * read(void * arg); void * accu(void * arg); static sem_t sem_one; static sem_t sem_two; static int num; int main(int argc, char *argv[]) { pthread_t id_t1, id_t2; sem_init(&sem_one, 0, 0); sem_init(&sem_two, 0, 1); pthread_create(&id_t1, NULL, read, NULL); pthread_create(&id_t2, NULL, accu, NULL); pthread_join(id_t1, NULL); pthread_join(id_t2, NULL); sem_destroy(&sem_one); sem_destroy(&sem_two); return 0; } void * read(void * arg) { int i; for (i = 0; i < 5; i++) { fputs("Input num: ", stdout); sem_wait(&sem_two); scanf("%d", &num); sem_post(&sem_one); } return NULL; } void * accu(void * arg) { int sum = 0 , i; for (i = 0; i < 5; i++) { sem_wait(&sem_one); sum+= num; sem_post(&sem_two); } printf("Result: %d \n", sum); return NULL; }

　　補(bǔ)充：線程的銷毀，線程創(chuàng)建后并不是其入口函數(shù)返回后就會(huì)自動(dòng)銷毀，需要手動(dòng)銷毀，不然線程創(chuàng)建的內(nèi)存空間將一直存在。一般手動(dòng)銷毀有如下兩種方式：1，調(diào)用pthread_join函數(shù)，其返回后同時(shí)銷毀線程 ，是一個(gè)阻斷函數(shù)，服務(wù)端一般不用它銷毀，因?yàn)榉?wù)端主線程不宜阻斷，還要實(shí)時(shí)監(jiān)聽客服端連接。2，調(diào)用pthread_detach函數(shù)，不會(huì)阻塞，線程返回自動(dòng)銷毀線程，不過要注意調(diào)用它后不能再調(diào)用pthread_join函數(shù)，它與pthread_join主要區(qū)別就是一個(gè)是阻塞函數(shù)，一個(gè)不阻塞。

　　多線程并發(fā)服務(wù)端的實(shí)現(xiàn)

　　使用多線程實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的聊天程序，并對(duì)臨界區(qū)(clnt_cnt,clnt_socks)進(jìn)行加鎖訪問.

　　服務(wù)端：

　　// // main.cpp // hello_server // // Created by app05 on 15-10-22. // Copyright (c) 2015年 app05. All rights reserved. //臨界區(qū)是：clnt_cnt和clnt_socks訪問處 #include #include #include #include #include #include #include #define BUF_SIZE 100 #define MAX_CLNT 256 void * handle_clnt(void * arg); void send_msg(char *msg, int len); void error_handling(char * msg); int clnt_cnt = 0; int clnt_socks[MAX_CLNT]; pthread_mutex_t mutx; int main(int argc, char *argv[]) { int serv_sock, clnt_sock; struct sockaddr_in serv_adr, clnt_adr; socklen_t clnt_adr_sz; pthread_t t_id; if (argc != 2) { printf("Usage : %s \n", argv[0]); exit(1); } //創(chuàng)建互斥量 pthread_mutex_init(&mutx, NULL); serv_sock = socket(PF_INET, SOCK_STREAM, 0); memset(&serv_adr, 0, sizeof(serv_adr)); serv_adr.sin_family = AF_INET; serv_adr.sin_addr.s_addr = htonl(INADDR_ANY); serv_adr.sin_port = htons(atoi(argv[1])); if(bind(serv_sock, (struct sockaddr *) &serv_adr, sizeof(serv_adr)) == -1) error_handling("bind() error"); if(listen(serv_sock, 5) == -1) error_handling("listen() error"); while (1) { clnt_adr_sz = sizeof(clnt_adr); clnt_sock = accept(serv_sock, (struct sockaddr*)&clnt_adr, &clnt_adr_sz); //阻斷，監(jiān)聽客服端連接請(qǐng)求 //臨界區(qū) pthread_mutex_lock(&mutx); //加鎖 clnt_socks[clnt_cnt++] = clnt_sock; //新連接的客服端保存到clnt_socks數(shù)組里 pthread_mutex_unlock(&mutx); //釋放鎖 //創(chuàng)建線程 pthread_create(&t_id, NULL, handle_clnt, (void*) &clnt_sock); pthread_detach(t_id); //銷毀線程，線程return后自動(dòng)調(diào)用銷毀，不阻斷 printf("Connected client IP: %s \n", inet_ntoa(clnt_adr.sin_addr)); } close(serv_sock); return 0; } //線程執(zhí)行 void * handle_clnt(void * arg) { int clnt_sock = *((int *)arg); int str_len = 0, i; char msg[BUF_SIZE]; while ((str_len = read(clnt_sock, msg, sizeof(msg))) != 0) send_msg(msg, str_len); //從數(shù)組中移除當(dāng)前客服端 pthread_mutex_lock(&mutx); for (i = 0; i < clnt_cnt; i++) { if (clnt_sock == clnt_socks[i]) { while (i++ < clnt_cnt - 1) clnt_socks[i] = clnt_socks[i + 1]; break; } } clnt_cnt--; pthread_mutex_unlock(&mutx); close(clnt_sock); return NULL; } //向所有連接的客服端發(fā)送消息 void send_msg(char * msg, int len) { int i; pthread_mutex_lock(&mutx); for (i = 0; i < clnt_cnt; i++) write(clnt_socks[i], msg, len); pthread_mutex_unlock(&mutx); } void error_handling(char *message) { fputs(message, stderr); fputc('\n', stderr); exit(1); }

　　客服端

　　// // main.cpp // hello_client // // Created by app05 on 15-10-22. // Copyright (c) 2015年 app05. All rights reserved. // // #include #include #include #include #include #include #include #define BUF_SIZE 100 #define NAME_SIZE 20 void * send_msg(void * arg); void * recv_msg(void * arg); void error_handling(char *message); char name[NAME_SIZE] = "[DEFAULT]"; char msg[BUF_SIZE]; int main(int argc, const char * argv[]) { int sock; struct sockaddr_in serv_addr; pthread_t snd_thread, rcv_thread; void * thread_return; if(argc != 4) { printf("Usage: %s \n", argv[0]); exit(1); } sprintf(name, "[%s]", argv[3]); //聊天人名字，配置到編譯器參數(shù)里 sock = socket(PF_INET, SOCK_STREAM, 0); if(sock == -1) error_handling("socket() error"); memset(&serv_addr, 0, sizeof(serv_addr)); serv_addr.sin_family = AF_INET; serv_addr.sin_addr.s_addr = inet_addr(argv[1]); serv_addr.sin_port = htons(atoi(argv[2])); if (connect(sock, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) == -1) error_handling("connect() error"); //多線程分離輸入和輸出 （北京it外包公司）pthread_create(&snd_thread, NULL, send_msg, (void *)&sock); pthread_create(&rcv_thread, NULL, recv_msg, (void *)&sock); //阻塞，等待返回 pthread_join(snd_thread, &thread_return); pthread_join(rcv_thread, &thread_return); close(sock); return 0; } //發(fā)送消息 void * send_msg(void * arg) { int sock = *((int *)arg); char name_msg[NAME_SIZE + BUF_SIZE]; while (1) { fgets(msg, BUF_SIZE, stdin); if (!strcmp(msg, "q\n") || !strcmp(msg, "Q \n")) { close(sock); exit(0); } sprintf(name_msg, "%s %s", name, msg); write(sock, name_msg, strlen(name_msg)); } return NULL; } //接收消息 void * recv_msg(void * arg) { int sock = *((int *)arg); char name_msg[NAME_SIZE + BUF_SIZE]; int str_len; while (1) { str_len = read(sock, name_msg, NAME_SIZE + BUF_SIZE - 1); if(str_len == -1) return (void *)-1; name_msg[str_len] = 0; fputs(name_msg, stdout); } return NULL; } void error_handling(char *message) { fputs(message, stderr); fputc('\n', stderr); exit(1); }

　　艾銻無限是中國(guó)領(lǐng)先IT外包服務(wù)商，專業(yè)為企業(yè)提供IT運(yùn)維外包、電腦維護(hù)、網(wǎng)絡(luò)維護(hù)、網(wǎng)絡(luò)布線、辦公設(shè)備維護(hù)、服務(wù)器維護(hù)、數(shù)據(jù)備份恢復(fù)、門禁監(jiān)控、網(wǎng)站建設(shè)等多項(xiàng)IT服務(wù)外包,服務(wù)熱線：400-650-7820 聯(lián)系電話：010-62684652 咨詢QQ1548853602 地址：北京市海淀區(qū)北京科技會(huì)展2號(hào)樓16D,用心服務(wù)每一天，為企業(yè)的發(fā)展提升更高的效率，創(chuàng)造更大的價(jià)值。

　　更多的IT外包信息盡在艾銻無限http://www.zhengyunzhengyun.cn