并发服务器的设计与实现

并发服务器

1.基于多线程的并发服务器

并发服务器支持多个客户端的连接，最大可接入的客户端数取决于内核控制块的个数。 当使用Socket API时，要使服务器能够同时支持多个客户端的连接，必须引入多任务机制，为每个连接创建一个单独的任务来处理连接上的数据，我们将这个设计方式称作并发服务器的设计。

由于多线程并发服务器涉及到子任务的动态创建和销毁，用户需要自己完成对任务堆栈的管理和回收，因此并发服务器的设计流程也相对复杂。

以下并发服务器实例完成的功能为：服务器能够同时支持多个客户端的连接，并能够将每个连接上接收到的小写字母转换成大写字母回显到客户端，其实现步骤如下

参考Socket API编程优化一文，在该文的工程源码基础上进行修改

在工程中创建socket_thread_server.c和对应的头文件

/******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){
  // 每一个任务，都有独立的栈空间
  int cfd = * (int *)argument;
  int n, i;
  while(1){
    //等待客户端发送数据
    n = Read(cfd, ReadBuff, BUFF_SIZE);
    if(n <= 0){
      close(cfd);
      vTaskDelete(NULL);
    }
    //进行大小写转换
    for(i = 0; i < n; i++){	
      ReadBuff[i] = toupper(ReadBuff[i]);		
    }
    //写回客户端
    n = Write(cfd, ReadBuff, n);
    if(n < 0){
      close(cfd);
      vTaskDelete(NULL);			
    }
  }
}
/**
  * @brief  多线程服务器
  * @param  none
  * @retval none
  */
void vThreadServerTask(void){
  int sfd, cfd;
  struct sockaddr_in server_addr, client_addr;
  socklen_t	client_addr_len;
  //创建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));
  //监听socket
  Listen(sfd, 5);
  //等待客户端连接
  client_addr_len = sizeof(client_addr);
  while(1){
    /*每创建一个socket，lwip都会分配一片内存空间
    宏NUM_SOCKETS就定义了一共支持多少个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文件中的默认任务里面添加代码

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 */
}

编译无误下载到开发板后，打开串口助手可以看到相关调试信息，使用网络调试工具可以创建多个PC客户端（串口会返回对应的cfd），输入任意小写字母，Server将返回对应的大写字母

2.基于Select的并发服务器

基于多线程的socket并发服务器，必须使用多线程的方式来实现，即为每个连接创建一个单独的任务来处理数据。 但是，这种多线程的方式是有缺陷的，在大型服务器的设计中，一个服务器上可能存在成千上万条连接，如果为每个连接都创建一个线程，这对系统资源来说无疑是比巨大的开销，也是种不太现实的做法。 事实上，在socket编程中，通常使用一种叫做Select的机制来实现并发服务器的设计。

Select函数实现的基本思想为：先构造一张有关描述符的表，然后调用一个函数。 当这些文件描述符中的一个或多个已准备好进行I/O时函数才返回； 函数返回时告诉进程哪个描述符已就绪，可以进行I/O操作

/*****select()函数*****/
函数原型：int select(int maxfd, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)
传 入 值：maxfd 监控的文件描述符集里最大文件描述符加1
	readfds 监控有读数据到达文件描述符集合，传入传出参数
	writefds 监控有写数据到达文件描述符集合，传入传出参数
	exceptfds 监控异常发生达文件描述符集合，传入传出参数
	timeout 超时设置 
	-->NULL：一直阻塞，直到有文件描述符就绪或出错
	-->0：仅仅检测文件描述符集的状态，然后立即返回，轮询
	-->不为0：在指定时间内，如果没有事件发生，则超时返回
返 回 值：成功：所监听的所有监听集合中，满足条件的总数！
	失败：0 超时
	错误：-1
//timeval结构体
struct timeval {
    long tv_sec; /* seconds */
    long tv_usec; /* microseconds */
};

调用 select() 函数时进程会一直阻塞直到有文件可读、有文件可写或者超时时间到。 为了设置文件描述符需要使用几个宏：

• select能监听的文件描述符个数受限于FD_SETSIZE,一般为1024，单纯改变进程打开的文件描述符个数并不能改变select监听文件个数
• 解决1024以下客户端时使用select是很合适的，但如果链接客户端过多，select采用的是轮询模型，会大大降低服务器响应效率，不应在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)测试给定的位置是否置位。
if(FD_ISSET(fd,&rset))
{......}

select编程模型如下图示

以下并发服务器实例完成的功能为：服务器能够同时支持多个客户端的连接，并能够将每个连接上接收到的小写字母转换成大写字母回显到客户端，其实现步骤如下：

参考Socket API编程优化一文，在该文的工程源码基础上进行修改

在工程中创建socket_socket_server.c和对应的头文件

#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 并发服务器
  * @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里面包含了服务器的fd
  int clientfds[FD_SETSIZE - 1];	
  //创建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));
  //监听socket
  Listen(sfd, 5);	
  client_addr_len = sizeof(client_addr);
  //初始化 maxfd 等于 sfd
  maxfd = sfd;	
  //清空fdset
  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集合就会变化，再select时，就不能使用，
    //所以我们要保存设置fd_set 和 读取的fd_set
    read_set = all_set;
    nready = select(maxfd + 1, &read_set, NULL, NULL, NULL);
    //没有超时机制，不会返回0
    if(nready < 0){
      printf("select error \\r\\n");
      vTaskDelete(NULL);
    }
    //判断监听的套接字是否有数据
    if(FD_ISSET(sfd, &read_set)){	
      //有客户端进行连接了
      cfd = accept(sfd, (struct sockaddr *)&client_addr, &client_addr_len);
      if(cfd < 0){
        printf("accept socket error\\r\\n");
        //继续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;		
        }
      }
      //没有其他套接字需要处理：这里防止重复工作，就不去执行其他任务
      if(--nready == 0){
        //继续select
        continue;
      }	
    }
    //遍历所有的客户端文件描述符
    for(i = 0; i < FD_SETSIZE -1 ; i++){
      if(clientfds[i] == -1){
        //继续遍历
        continue;
      }
      //是否在我们fd_set集合里面
      if(FD_ISSET(clientfds[i], &read_set)){
        n = Read(clientfds[i], ReadBuff, BUFF_SIZE);
        //Read函数已经关闭了这个客户端的fd
        if(n <= 0){
          //从集合里面清除
          FD_CLR(clientfds[i], &all_set);
          //当前的客户端fd 赋值为-1
          clientfds[i] = -1;
        }else{
          //进行大小写转换
          for(j = 0; j < n; j++){		
            ReadBuff[j] = toupper(ReadBuff[j]);		
          }
          //写回客户端
          n = Write(clientfds[i], ReadBuff, n);
          if(n < 0){
            //从集合里面清除
            FD_CLR(clientfds[i], &all_set);
            //当前的客户端fd 赋值为-1
            clientfds[i] = -1;		
          }				
        }
      }
    }		
  }
}

在freertos.c文件中的默认任务里面添加代码

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 */
}

编译无误下载到开发板后，打开串口助手可以看到相关调试信息，使用网络调试工具可以创建多个PC客户端（串口会返回对应的cfd），输入任意小写字母，Server将返回对应的大写字母