TCP/IP網絡編程多線程服務器端的實現

作者：夜色魅影 2015-10-27 09:40:31

前面我們講過多進程服務器，但我們知道它開銷很大，因此我們才引入線程，我們可以把它看成是一種輕量級進程。它相比進程有如下幾個優點：線程的創建和上下文切換開銷更小且速度更快。線程間交換數據時無需特殊技術。進程：在操作系統構成單獨執行流的單位。線程：在進程構成單獨執行流的單位。

線程基本概念

前面我們講過多進程服務器，但我們知道它開銷很大，因此我們才引入線程，我們可以把它看成是一種輕量級進程。它相比進程有如下幾個優點：

線程的創建和上下文切換開銷更小且速度更快。

線程間交換數據時無需特殊技術。

進程：在操作系統構成單獨執行流的單位。

線程：在進程構成單獨執行流的單位。

它們的包含關系是，操作系統 > 進程 > 線程。進程與線程具體差異其實是這樣的，每個進程都有獨立的完整內存空間，它包括全局數據區，堆區，棧區，而多進程服務器之所以開銷大是因為只是為了區分棧區里的不同函數流執行而把數據區，堆區，棧區內存全部復制了一份。而多線程就高效多了，它只把棧區分離出來，進程中的數據區，堆區則共享。具體內存結構示例圖如下：

線程創建及運行

線程具有單獨的執行流，因此需要單獨定義線程的入口函數，而且還需要請求操作系統在單獨的執行流中執行該函數，完成這個功能的函數如下：

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

實例代碼：

#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; 
 
}

上面實例是用sleep延遲來控制線程的執行的，如果主線程不做延遲那么執行到return 0;時，進程就結束了，相應的線程也會銷毀。而明顯用sleep這種方式控制線程執行流是不合理的，下面我們來看看一個更好的延遲函數，調用該函數的進程(或線程)將進入等待狀態，直到***個參數為ID的線程終止為止。而且可以得到線程的入口函數返回值。

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

實例代碼：

#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; 
 
//創建線程 
 
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; 
 
} 
 
//線程入口函數 
 
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; 
 
}

#p#

線程存在的問題和臨界區

前面我們知道了怎么創建線程，但我們都是只創建了一個線程，下面我們再來看看這樣一個實例，創建100個線程，它們都訪問了同一變量，其中一半對這個變量進行加1操作，一半進行減1操作，按道理其結果會等于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位整數型，多線程共同訪問 
 
int main(int argc, char *argv[]) 
 
{ 
 
pthread_t thread_id[NUM_THREAD]; 
 
int i; 
 
//創建100個線程，一半執行thread_inc,一半執行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按道理結果是0 
 
return 0; 
 
} 
 
//線程入口函數1 
 
void * thread_inc(void * arg) 
 
{ 
 
for (int i = 0; i < 50000000; i++) 
 
num += 1;//臨界區(引起問題的語句就是臨界區位置) 
 
return NULL; 
 
} 
 
//線程入口函數2 
 
void * thread_des(void * arg) 
 
{ 
 
for (int i = 0; i < 50000000; i++) 
 
num -= 1;//臨界區 
 
return NULL; 
 
}

從運行結果看并不是0，而且每次運行的結果都不同。那這是什么原因引起的呢? 是因為每個線程訪問一個變量是這樣一個過程：先從內存取出這個變量值到CPU，然后CPU計算得到改變后的值，***再將這個改變后的值寫回內存。因此，我們可以很容易看出，多個線程訪問同一變量，如果某個線程還只剛從內存取出數據，還沒來得及寫回內存，這時其它線程又訪問了這個變量，所以這個值就會不正確了。

接下來我們再來講講怎么解決這個問題：線程同步

線程同步

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

同時訪問同一內存空間時發生的情況

需要指定訪問同一內存空間的線程執行順序的情況

針對這兩種可能引發的情況，我們分別使用的同步技術是：互斥量和信號量。

互斥量

互斥量技術從字面也可以理解，就是臨界區有線程訪問，其它線程就得排隊等待，它們的訪問是互斥的，實現方式就是給臨界區加鎖與釋放鎖。

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

簡言之，就是利用lock和unlock函數圍住臨界區的兩端。當某個線程調用pthread_mutex_lock進入臨界區后，如果沒有調用pthread_mutex_unlock釋放鎖退出，那么其它線程就會一直阻塞在臨界區之外，我們把這種情況稱之為死鎖。所以臨界區圍住一定要lock和unlock一一對應。

實例代碼：

#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; 
 
//互斥量的創建 
 
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; 
 
} 
 
/*擴展臨界區，減少加鎖，釋放鎖調用次數，但這樣變量必須加滿到50000000次后其它線程才能訪問. 
 
這樣是延長了線程的等待時間，但縮短了加鎖，釋放鎖函數調用的時間，這里沒有定論，自己酌情考慮*/ 
 
void * thread_inc(void * arg) 
 
{ 
 
pthread_mutex_lock(&mutex); //互斥量鎖住 
 
for (int i = 0; i < 1000000; i++) 
 
num += 1; 
 
pthread_mutex_unlock(&mutex); //互斥量釋放鎖 
 
return NULL; 
 
} 
 
/*縮短了線程等待時間，但循環創建，釋放鎖函數調用時間增加*/ 
 
void * thread_des(void * arg) 
 
{ 
 
for (int i = 0; i < 1000000; i++) 
 
{ 
 
pthread_mutex_lock(&mutex); 
 
num -= 1; 
 
pthread_mutex_unlock(&mutex); 
 
} 
 
return NULL; 
 
}

#p#

信號量

信號量與互斥量類似，只是互斥量是用鎖來控制線程訪問而信號量是用二進制0，1來完成控制線程順序。sem_post信號量加1，sem_wait信號量減1，當信號量為0時，sem_wait就會阻斷，因此通過這樣讓信號量加1減1就能控制線程的執行順序了。

注釋：mac上測試信號量函數返回-1失敗，以后還是Linux上整吧，也許這些接口已經過時了…

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

實例代碼：線程A從用戶輸入得到值后存入全局變量num，此時線程B將取走該值并累加。該過程共進行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; 
 
}

補充：線程的銷毀，線程創建后并不是其入口函數返回后就會自動銷毀，需要手動銷毀，不然線程創建的內存空間將一直存在。一般手動銷毀有如下兩種方式：1，調用pthread_join函數，其返回后同時銷毀線程，是一個阻斷函數，服務端一般不用它銷毀，因為服務端主線程不宜阻斷，還要實時監聽客服端連接。2，調用pthread_detach函數，不會阻塞，線程返回自動銷毀線程，不過要注意調用它后不能再調用pthread_join函數，它與pthread_join主要區別就是一個是阻塞函數，一個不阻塞。

多線程并發服務端的實現

使用多線程實現了一個簡單的聊天程序，并對臨界區(clnt_cnt,clnt_socks)進行加鎖訪問.

服務端：

// 
 
// main.cpp 
 
// hello_server 
 
// 
 
// Created by app05 on 15-10-22. 
 
// Copyright (c) 2015年 app05. All rights reserved. 
 
//臨界區是：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); 
 
} 
 
//創建互斥量 
 
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); //阻斷，監聽客服端連接請求 
 
//臨界區 
 
pthread_mutex_lock(&mutx); //加鎖 
 
clnt_socks[clnt_cnt++] = clnt_sock; //新連接的客服端保存到clnt_socks數組里 
 
pthread_mutex_unlock(&mutx); //釋放鎖 
 
//創建線程 
 
pthread_create(&t_id, NULL, handle_clnt, (void*) &clnt_sock); 
 
pthread_detach(t_id); //銷毀線程，線程return后自動調用銷毀，不阻斷 
 
printf("Connected client IP: %s \n", inet_ntoa(clnt_adr.sin_addr)); 
 
} 
 
close(serv_sock); 
 
return 0; 
 
} 
 
//線程執行 
 
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); 
 
//從數組中移除當前客服端 
 
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; 
 
} 
 
//向所有連接的客服端發送消息 
 
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); 
 
}

#p#

客服端

// 
 
// 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]); //聊天人名字，配置到編譯器參數里 
 
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"); 
 
//多線程分離輸入和輸出 
 
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; 
 
} 
 
//發送消息 
 
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); 
 
}

責任編輯：何妍來源： CSDN博客

TCP IP 網絡協議線程

TCP/IP網絡編程 多線程服務器端的實現

TCP/IP網絡編程多線程服務器端的實現