并發(fā)服務(wù)器的設(shè)計(jì)與實(shí)現(xiàn)

并發(fā)服務(wù)器

1.基于多線程的并發(fā)服務(wù)器

并發(fā)服務(wù)器支持多個(gè)客戶端的連接，最大可接入的客戶端數(shù)取決于內(nèi)核控制塊的個(gè)數(shù)。 當(dāng)使用Socket API時(shí)，要使服務(wù)器能夠同時(shí)支持多個(gè)客戶端的連接，必須引入多任務(wù)機(jī)制，為每個(gè)連接創(chuàng)建一個(gè)單獨(dú)的任務(wù)來(lái)處理連接上的數(shù)據(jù)，我們將這個(gè)設(shè)計(jì)方式稱作并發(fā)服務(wù)器的設(shè)計(jì)。

由于多線程并發(fā)服務(wù)器涉及到子任務(wù)的動(dòng)態(tài)創(chuàng)建和銷(xiāo)毀，用戶需要自己完成對(duì)任務(wù)堆棧的管理和回收，因此并發(fā)服務(wù)器的設(shè)計(jì)流程也相對(duì)復(fù)雜。

以下并發(fā)服務(wù)器實(shí)例完成的功能為：服務(wù)器能夠同時(shí)支持多個(gè)客戶端的連接，并能夠?qū)⒚總€(gè)連接上接收到的小寫(xiě)字母轉(zhuǎn)換成大寫(xiě)字母回顯到客戶端，其實(shí)現(xiàn)步驟如下

參考Socket API編程優(yōu)化一文，在該文的工程源碼基礎(chǔ)上進(jìn)行修改

在工程中創(chuàng)建socket_thread_server.c和對(duì)應(yīng)的頭文件

/******socket_thread_server.c******/
#include "socket_tcp_server.h"
#include "socket_wrap.h"
#include "FreeRTOS.h"
#include "task.h"
#include "cmsis_os.h"
#include "ctype.h"

static char ReadBuff[BUFF_SIZE];
/**
  * @brief  Function implementing the defaultTask thread.
  * @param  argument: Not used 
  * @retval None
  */
/* USER CODE END Header_StartDefaultTask */
void vNewClientTask(void const * argument){
  // 每一個(gè)任務(wù)，都有獨(dú)立的?？臻g
  int cfd = * (int *)argument;
  int n, i;
  while(1){
    //等待客戶端發(fā)送數(shù)據(jù)
    n = Read(cfd, ReadBuff, BUFF_SIZE);
    if(n <= 0){
      close(cfd);
      vTaskDelete(NULL);
    }
    //進(jìn)行大小寫(xiě)轉(zhuǎn)換
    for(i = 0; i < n; i++){	
      ReadBuff[i] = toupper(ReadBuff[i]);		
    }
    //寫(xiě)回客戶端
    n = Write(cfd, ReadBuff, n);
    if(n < 0){
      close(cfd);
      vTaskDelete(NULL);			
    }
  }
}
/**
  * @brief  多線程服務(wù)器
  * @param  none
  * @retval none
  */
void vThreadServerTask(void){
  int sfd, cfd;
  struct sockaddr_in server_addr, client_addr;
  socklen_t	client_addr_len;
  //創(chuàng)建socket
  sfd = Socket(AF_INET, SOCK_STREAM, 0);
  server_addr.sin_family = AF_INET;
  server_addr.sin_port   = htons(SERVER_PORT);
  server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
  //綁定socket
  Bind(sfd, (struct sockaddr *)&server_addr, sizeof(server_addr));
  //監(jiān)聽(tīng)socket
  Listen(sfd, 5);
  //等待客戶端連接
  client_addr_len = sizeof(client_addr);
  while(1){
    /*每創(chuàng)建一個(gè)socket，lwip都會(huì)分配一片內(nèi)存空間
    宏NUM_SOCKETS就定義了一共支持多少個(gè)socket,即能分配多少fd
    #define NUM_SOCKETS		MEMP_NUM_NETCONN
    #define MEMP_NUM_NETCONN	8		
    */
    cfd = Accept(sfd,(struct sockaddr *)&client_addr, &client_addr_len);
    printf("client is connect cfd = %d\\r\\n",cfd);
    if(xTaskCreate((TaskFunction_t) vNewClientTask,
		   "Client",
		   128,//1k
		   (void *)&cfd,
		   osPriorityNormal,
		   NULL) != pdPASS){	
      printf("create task fail!\\r\\n");		
    }
  }									
}

在freertos.c文件中的默認(rèn)任務(wù)里面添加代碼

void StartDefaultTask(void const * argument){
  /* init code for LWIP */
  MX_LWIP_Init();
  /* USER CODE BEGIN StartDefaultTask */
  printf("TCP thread server started!\\r\\n",cfd);
  /* Infinite loop */
  for(;;){
    vThreadServerTask();
    osDelay(100);
  }
  /* USER CODE END StartDefaultTask */
}

編譯無(wú)誤下載到開(kāi)發(fā)板后，打開(kāi)串口助手可以看到相關(guān)調(diào)試信息，使用網(wǎng)絡(luò)調(diào)試工具可以創(chuàng)建多個(gè)PC客戶端（串口會(huì)返回對(duì)應(yīng)的cfd），輸入任意小寫(xiě)字母，Server將返回對(duì)應(yīng)的大寫(xiě)字母

2.基于Select的并發(fā)服務(wù)器

基于多線程的socket并發(fā)服務(wù)器，必須使用多線程的方式來(lái)實(shí)現(xiàn)，即為每個(gè)連接創(chuàng)建一個(gè)單獨(dú)的任務(wù)來(lái)處理數(shù)據(jù)。 但是，這種多線程的方式是有缺陷的，在大型服務(wù)器的設(shè)計(jì)中，一個(gè)服務(wù)器上可能存在成千上萬(wàn)條連接，如果為每個(gè)連接都創(chuàng)建一個(gè)線程，這對(duì)系統(tǒng)資源來(lái)說(shuō)無(wú)疑是比巨大的開(kāi)銷(xiāo)，也是種不太現(xiàn)實(shí)的做法。 事實(shí)上，在socket編程中，通常使用一種叫做Select的機(jī)制來(lái)實(shí)現(xiàn)并發(fā)服務(wù)器的設(shè)計(jì)。

Select函數(shù)實(shí)現(xiàn)的基本思想為：先構(gòu)造一張有關(guān)描述符的表，然后調(diào)用一個(gè)函數(shù)。 當(dāng)這些文件描述符中的一個(gè)或多個(gè)已準(zhǔn)備好進(jìn)行I/O時(shí)函數(shù)才返回； 函數(shù)返回時(shí)告訴進(jìn)程哪個(gè)描述符已就緒，可以進(jìn)行I/O操作

/*****select()函數(shù)*****/
函數(shù)原型：int select(int maxfd, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)
傳 入 值：maxfd 監(jiān)控的文件描述符集里最大文件描述符加1
	readfds 監(jiān)控有讀數(shù)據(jù)到達(dá)文件描述符集合，傳入傳出參數(shù)
	writefds 監(jiān)控有寫(xiě)數(shù)據(jù)到達(dá)文件描述符集合，傳入傳出參數(shù)
	exceptfds 監(jiān)控異常發(fā)生達(dá)文件描述符集合，傳入傳出參數(shù)
	timeout 超時(shí)設(shè)置 
	-->NULL：一直阻塞，直到有文件描述符就緒或出錯(cuò)
	-->0：僅僅檢測(cè)文件描述符集的狀態(tài)，然后立即返回，輪詢
	-->不為0：在指定時(shí)間內(nèi)，如果沒(méi)有事件發(fā)生，則超時(shí)返回
返 回 值：成功：所監(jiān)聽(tīng)的所有監(jiān)聽(tīng)集合中，滿足條件的總數(shù)！
	失?。? 超時(shí)
	錯(cuò)誤：-1
//timeval結(jié)構(gòu)體
struct timeval {
    long tv_sec; /* seconds */
    long tv_usec; /* microseconds */
};

調(diào)用 select() 函數(shù)時(shí)進(jìn)程會(huì)一直阻塞直到有文件可讀、有文件可寫(xiě)或者超時(shí)時(shí)間到。 為了設(shè)置文件描述符需要使用幾個(gè)宏：

• select能監(jiān)聽(tīng)的文件描述符個(gè)數(shù)受限于FD_SETSIZE,一般為1024，單純改變進(jìn)程打開(kāi)的文件描述符個(gè)數(shù)并不能改變select監(jiān)聽(tīng)文件個(gè)數(shù)
• 解決1024以下客戶端時(shí)使用select是很合適的，但如果鏈接客戶端過(guò)多，select采用的是輪詢模型，會(huì)大大降低服務(wù)器響應(yīng)效率，不應(yīng)在select上投入更多精力

#include 
int FD_ZERO(fd_set *fdset);		//從fdset中清除所有的文件描述符
int FD_CLR(int fd,fd_set *fdset);	//將fd從fdset中清除
int FD_SET(int fd,fd_set *fdset);	//將fd加入到fdset
int FD_ISSET(int fd,fd_set *fdset);	//判斷fd是否在fdset集合中
/*例如*/
fd_set rset;
int fd;
FD_ZERO(&rset);
FD_SET(fd,&rset);
FD_SET(stdin，&rset);
//在select返回之后，可以使用FD_ISSET(fd,&rset)測(cè)試給定的位置是否置位。
if(FD_ISSET(fd,&rset))
{......}

select編程模型如下圖示

以下并發(fā)服務(wù)器實(shí)例完成的功能為：服務(wù)器能夠同時(shí)支持多個(gè)客戶端的連接，并能夠?qū)⒚總€(gè)連接上接收到的小寫(xiě)字母轉(zhuǎn)換成大寫(xiě)字母回顯到客戶端，其實(shí)現(xiàn)步驟如下：

參考Socket API編程優(yōu)化一文，在該文的工程源碼基礎(chǔ)上進(jìn)行修改

在工程中創(chuàng)建socket_socket_server.c和對(duì)應(yīng)的頭文件

#include "socket_wrap.h"
#include "socket_select_server.h"
#include "socket_tcp_server.h"
#include "string.h"
#include "FreeRTOS.h"
#include "task.h"
#include "ctype.h"

static char ReadBuff[BUFF_SIZE];
/**
  * @brief  select 并發(fā)服務(wù)器
  * @param  none
  * @retval none
  */
void vSelectServerTask(void){
  int sfd, cfd, maxfd, i, nready, n, j;
  struct sockaddr_in server_addr, client_addr;
  socklen_t	client_addr_len;
  fd_set all_set, read_set;
  //FD_SETSIZE里面包含了服務(wù)器的fd
  int clientfds[FD_SETSIZE - 1];	
  //創(chuàng)建socket
  sfd = Socket(AF_INET, SOCK_STREAM, 0);
  server_addr.sin_family = AF_INET;
  server_addr.sin_port = htons(SERVER_PORT);
  server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
  //綁定socket
  Bind(sfd, (struct sockaddr *)&server_addr, sizeof(server_addr));
  //監(jiān)聽(tīng)socket
  Listen(sfd, 5);	
  client_addr_len = sizeof(client_addr);
  //初始化 maxfd 等于 sfd
  maxfd = sfd;	
  //清空f(shuō)dset
  FD_ZERO(&all_set);	
  //把sfd文件描述符添加到集合中	
  FD_SET(sfd, &all_set);
  //初始化客戶端fd的集合
  for(i = 0; i < FD_SETSIZE -1 ; i++){
    //初始化為-1
    clientfds[i] = -1;
  }
  while(1){
    //每次select返回之后，fd_set集合就會(huì)變化，再select時(shí)，就不能使用，
    //所以我們要保存設(shè)置fd_set 和 讀取的fd_set
    read_set = all_set;
    nready = select(maxfd + 1, &read_set, NULL, NULL, NULL);
    //沒(méi)有超時(shí)機(jī)制，不會(huì)返回0
    if(nready < 0){
      printf("select error \\r\\n");
      vTaskDelete(NULL);
    }
    //判斷監(jiān)聽(tīng)的套接字是否有數(shù)據(jù)
    if(FD_ISSET(sfd, &read_set)){	
      //有客戶端進(jìn)行連接了
      cfd = accept(sfd, (struct sockaddr *)&client_addr, &client_addr_len);
      if(cfd < 0){
        printf("accept socket error\\r\\n");
        //繼續(xù)select
        continue;
      }
      printf("new client connect fd = %d\\r\\n", cfd);
      //把新的cfd 添加到fd_set集合中
      FD_SET(cfd, &all_set);
      //更新要select的maxfd
      maxfd = (cfd > maxfd)?cfd:maxfd;
      //把新的cfd 保存到cfds集合中
      for(i = 0; i < FD_SETSIZE -1 ; i++){
        if(clientfds[i] == -1){
          clientfds[i] = cfd;
          //退出，不需要添加
          break;		
        }
      }
      //沒(méi)有其他套接字需要處理：這里防止重復(fù)工作，就不去執(zhí)行其他任務(wù)
      if(--nready == 0){
        //繼續(xù)select
        continue;
      }	
    }
    //遍歷所有的客戶端文件描述符
    for(i = 0; i < FD_SETSIZE -1 ; i++){
      if(clientfds[i] == -1){
        //繼續(xù)遍歷
        continue;
      }
      //是否在我們fd_set集合里面
      if(FD_ISSET(clientfds[i], &read_set)){
        n = Read(clientfds[i], ReadBuff, BUFF_SIZE);
        //Read函數(shù)已經(jīng)關(guān)閉了這個(gè)客戶端的fd
        if(n <= 0){
          //從集合里面清除
          FD_CLR(clientfds[i], &all_set);
          //當(dāng)前的客戶端fd 賦值為-1
          clientfds[i] = -1;
        }else{
          //進(jìn)行大小寫(xiě)轉(zhuǎn)換
          for(j = 0; j < n; j++){		
            ReadBuff[j] = toupper(ReadBuff[j]);		
          }
          //寫(xiě)回客戶端
          n = Write(clientfds[i], ReadBuff, n);
          if(n < 0){
            //從集合里面清除
            FD_CLR(clientfds[i], &all_set);
            //當(dāng)前的客戶端fd 賦值為-1
            clientfds[i] = -1;		
          }				
        }
      }
    }		
  }
}

在freertos.c文件中的默認(rèn)任務(wù)里面添加代碼

void StartDefaultTask(void const * argument){
  /* init code for LWIP */
  MX_LWIP_Init();
  /* USER CODE BEGIN StartDefaultTask */
  printf("TCP thread server started!\\r\\n",cfd);
  /* Infinite loop */
  for(;;){
    vSocketServerTask();
    osDelay(100);
  }
  /* USER CODE END StartDefaultTask */
}

編譯無(wú)誤下載到開(kāi)發(fā)板后，打開(kāi)串口助手可以看到相關(guān)調(diào)試信息，使用網(wǎng)絡(luò)調(diào)試工具可以創(chuàng)建多個(gè)PC客戶端（串口會(huì)返回對(duì)應(yīng)的cfd），輸入任意小寫(xiě)字母，Server將返回對(duì)應(yīng)的大寫(xiě)字母