socket编程的同步、异步与阻塞、非阻塞示例详解

 socket编程的同步、异步与阻塞、非阻塞示例详解之一 

分类： 架构设计与优化

简介
图 1. 基本 Linux I/O 模型的简单矩阵
 
每个 I/O 模型都有自己的使用模式，它们对于特定的应用程序都有自己的优点。
本节将简要对其一一进行介绍。

一、同步阻塞模式
在这个模式中，用户空间的应用程序执行一个系统调用，并阻塞，直到系统调用完成为止(数据传输完成或发生错误)。

/*
 * \brief
 * tcp client
 */

#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netdb.h>
#include <string.h>
#define SERVPORT 8080
#define MAXDATASIZE 100

int main(int argc, char *argv[])
{
  int sockfd, recvbytes;
  char rcv_buf[MAXDATASIZE]; /*./client 127.0.0.1 hello */
  char snd_buf[MAXDATASIZE];
  struct hostent *host;             /* struct hostent
                                     * {
                                     * char *h_name; // general hostname
                                     * char **h_aliases; // hostname's alias
                                     * int h_addrtype; // AF_INET
                                     * int h_length;
                                     * char **h_addr_list;
                                     * };
                                     */
  struct sockaddr_in server_addr;

  if (argc < )
  {
    printf("Usage:%s [ip address] [any string]\n", argv[]);
    return ;
  }

  *snd_buf = '\0';
  strcat(snd_buf, argv[]);

  if ((sockfd = socket(AF_INET, SOCK_STREAM, )) == -)
  {
    perror("socket:");
    exit();
  }

  server_addr.sin_family = AF_INET;
  server_addr.sin_port = htons(SERVPORT);
  inet_pton(AF_INET, argv[], &server_addr.sin_addr);
  memset(&(server_addr.sin_zero), , );

  /* create the connection by socket
   * means that connect "sockfd" to "server_addr"
   * 同步阻塞模式
   */
  if (connect(sockfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -)
  {
    perror("connect");
    exit();
  }

  /* 同步阻塞模式  */
  if (send(sockfd, snd_buf, sizeof(snd_buf), ) == -)
  {
    perror("send:");
    exit();
  }
  printf("send:%s\n", snd_buf);

   /* 同步阻塞模式  */
  if ((recvbytes = recv(sockfd, rcv_buf, MAXDATASIZE, )) == -)
  {
    perror("recv:");
    exit();
  }

  rcv_buf[recvbytes] = '\0';
  printf("recv:%s\n", rcv_buf);

  close(sockfd);
  return ;
}

显然，代码中的connect, send, recv都是同步阻塞工作模式，
在结果没有返回时，程序什么也不做，

二、同步非阻塞模式
同步阻塞 I/O 的一种效率稍低的变种是同步非阻塞 I/O。
在这种模型中，系统调用是以非阻塞的形式打开的。
这意味着 I/O 操作不会立即完成， 操作可能会返回一个错误代码，
说明这个命令不能立即满足（EAGAIN 或 EWOULDBLOCK），
非阻塞的实现是 I/O 命令可能并不会立即满足，需要应用程序调用许多次来等待操作完成。
这可能效率不高，
因为在很多情况下，当内核执行这个命令时，应用程序必须要进行忙碌等待，直到数据可用为止，
或者试图执行其他工作。
因为数据在内核中变为可用到用户调用 read 返回数据之间存在一定的间隔，这会导致整体数据吞吐量的降低。

/*
 * \brief
 * tcp client
 */

#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <errno.h>
#include <netdb.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>

#define SERVPORT 8080
#define MAXDATASIZE 100

int main(int argc, char *argv[])
{
  int sockfd, recvbytes;
  char rcv_buf[MAXDATASIZE]; /*./client 127.0.0.1 hello */
  char snd_buf[MAXDATASIZE];
  struct hostent *host;             /* struct hostent
                                     * {
                                     * char *h_name; // general hostname
                                     * char **h_aliases; // hostname's alias
                                     * int h_addrtype; // AF_INET
                                     * int h_length;
                                     * char **h_addr_list;
                                     * };
                                     */
  struct sockaddr_in server_addr;
  int flags;
  int addr_len;

  if (argc < )
  {
    printf("Usage:%s [ip address] [any string]\n", argv[]);
    return ;
  }

  *snd_buf = '\0';
  strcat(snd_buf, argv[]);

  if ((sockfd = socket(AF_INET, SOCK_STREAM, )) == -)
  {
    perror("socket:");
    exit();
  }

  server_addr.sin_family = AF_INET;
  server_addr.sin_port = htons(SERVPORT);
  inet_pton(AF_INET, argv[], &server_addr.sin_addr);
  memset(&(server_addr.sin_zero), , );
  addr_len = sizeof(struct sockaddr_in);

  /* Setting socket to nonblock */
  flags = fcntl(sockfd, F_GETFL, );
  fcntl(sockfd, flags|O_NONBLOCK);

  /* create the connection by socket
   * means that connect "sockfd" to "server_addr"
   * 同步阻塞模式
  */
  if (connect(sockfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -)
  {
    perror("connect");
    exit();
  }

  /* 同步非阻塞模式 */
  while (send(sockfd, snd_buf, sizeof(snd_buf), MSG_DONTWAIT) == -)
  {
    sleep();
    printf("sleep\n");
  }
  printf("send:%s\n", snd_buf);

  /* 同步非阻塞模式 */
  while ((recvbytes = recv(sockfd, rcv_buf, MAXDATASIZE, MSG_DONTWAIT)) == -)
  {
    sleep();
    printf("sleep\n");
  }

  rcv_buf[recvbytes] = '\0';
  printf("recv:%s\n", rcv_buf);

  close(sockfd);
  return ;
}

异步阻塞模式，异步非阻塞模式以及server端程序见本文的第二部分。
http://blog.chinaunix.net/uid-26000296-id-3755268.html

 三、异步阻塞模式
 另外一个阻塞解决方案是带有阻塞通知的非阻塞 I/O。
 在这种模型中，配置的是非阻塞 I/O，然后使用阻塞 select 系统调用来确定一个 I/O 描述符何时有操作。
 使 select 调用非常有趣的是它可以用来为多个描述符提供通知，而不仅仅为一个描述符提供通知。
 对于每个提示符来说，我们可以请求这个描述符可以写数据、有读数据可用以及是否发生错误的通知

 下面的C语言实现的例子，它从网络上接受数据写入一个文件中:
 /*
  * \brief
  * tcp client
  */

 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/socket.h>
 #include <sys/select.h>
 #include <sys/time.h>
 #include <netdb.h>
 #include <string.h>

 #include <sys/types.h>
 #include <sys/stat.h>
 #include <fcntl.h>
 #define SERVPORT 8080
 #define MAXDATASIZE 100
 #define TFILE "data_from_socket.txt"

 int main(int argc, char *argv[])
 {
   int sockfd, recvbytes;
   char rcv_buf[MAXDATASIZE]; /*./client 127.0.0.1 hello */
   char snd_buf[MAXDATASIZE];
   struct hostent *host;             /* struct hostent
                                      * {
                                      * char *h_name; // general hostname
                                      * char **h_aliases; // hostname's alias
                                      * int h_addrtype; // AF_INET
                                      * int h_length;
                                      * char **h_addr_list;
                                      * };
                                      */
   struct sockaddr_in server_addr;

   /* */
   fd_set readset, writeset;
   int check_timeval = ;
   struct timeval timeout={check_timeval,}; //阻塞式select, 等待1秒，1秒轮询
   int maxfd;
   int fp;
   int cir_count = ;
   int ret;

   if (argc < )
   {
     printf("Usage:%s [ip address] [any string]\n", argv[]);
     return ;
   }

   *snd_buf = '\0';
   strcat(snd_buf, argv[]);

   if ((fp = open(TFILE,O_WRONLY)) < )    //不是用fopen
   {
     perror("fopen:");
     exit();
   }

   if ((sockfd = socket(AF_INET, SOCK_STREAM, )) == -)
   {
     perror("socket:");
     exit();
   }

   server_addr.sin_family = AF_INET;
   server_addr.sin_port = htons(SERVPORT);
   inet_pton(AF_INET, argv[], &server_addr.sin_addr);
   memset(&(server_addr.sin_zero), , );

   /* create the connection by socket
    * means that connect "sockfd" to "server_addr"
    */
   if (connect(sockfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -)
   {
     perror("connect");
     exit();
   }

   /**/
   if (send(sockfd, snd_buf, sizeof(snd_buf), ) == -)
   {
     perror("send:");
     exit();
   }
   printf("send:%s\n", snd_buf);

   while ()
   {
     FD_ZERO(&readset);            //每次循环都要清空集合，否则不能检测描述符变化
     FD_SET(sockfd, &readset);     //添加描述符
     FD_ZERO(&writeset);
     FD_SET(fp,     &writeset);

     maxfd = sockfd > fp ? (sockfd+) : (fp+);    //描述符最大值加1

     ret = select(maxfd, &readset, NULL, NULL, NULL);   // 阻塞模式
     switch( ret)
     {
       case -:
         exit(-);
         break;
       case :
         break;
       default:
         if (FD_ISSET(sockfd, &readset))  //测试sock是否可读，即是否网络上有数据
         {
           recvbytes = recv(sockfd, rcv_buf, MAXDATASIZE, MSG_DONTWAIT);
           rcv_buf[recvbytes] = '\0';
           printf("recv:%s\n", rcv_buf);

           if (FD_ISSET(fp, &writeset))
           {
             write(fp, rcv_buf, strlen(rcv_buf));   // 不是用fwrite
           }
           goto end;
         }
     }
     cir_count++;
     printf("CNT : %d \n",cir_count);
   }

 end:
   close(fp);
   close(sockfd);

   return ;
 }

 perl实现:
 #! /usr/bin/perl
 ###############################################################################
 # \File
 #  tcp_client.pl
 # \Descript
 #  send message to server
 ###############################################################################
 use IO::Socket;
 use IO::Select;

 #hash to install IP Port
 %srv_info =(
 #"srv_ip"  => "61.184.93.197",
       "srv_ip"  => "192.168.1.73",
       "srv_port"=> "",
       );

 my $srv_addr = $srv_info{"srv_ip"};
 my $srv_port = $srv_info{"srv_port"};

 my $sock = IO::Socket::INET->new(
       PeerAddr => "$srv_addr",
       PeerPort => "$srv_port",
       Type     => SOCK_STREAM,
       Blocking => ,
 #     Timeout  => ,
       Proto    => "tcp")
 or die "Can not create socket connect. $@";

 $sock->send("Hello server!\n", ) or warn "send failed: $!, $@";
 $sock->autoflush();

 my $sel = IO::Select->new($sock);
 while(my @ready = $sel->can_read)
 {
   foreach my $fh(@ready)
   {
     if($fh == $sock)
     {
       while()
       {
         print $_;
       }
       $sel->remove($fh);
       close $fh;
     }
   }
 }
 $sock->close();

 四、异步非阻塞模式
 最后，异步非阻塞 I/O 模型是一种处理与 I/O 重叠(并行)进行的模型。
 以read系统调用为例
 steps：
 a. 调用read;
 b. read请求会立即返回，说明请求已经成功发起了。
 c. 在后台完成读操作这段时间内，应用程序可以执行其他处理操作。
 d. 当 read 的响应到达时，就会产生一个信号或执行一个基于线程的回调函数来完成这次 I/O 处理过程。

 /*
  * \brief
  * tcp client
  */

 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/socket.h>
 #include <sys/select.h>
 #include <sys/time.h>
 #include <netdb.h>
 #include <string.h>

 #include <sys/types.h>
 #include <sys/stat.h>
 #include <fcntl.h>
 #define SERVPORT 8080
 #define MAXDATASIZE 100
 #define TFILE "data_from_socket.txt"

 int main(int argc, char *argv[])
 {
   int sockfd, recvbytes;
   char rcv_buf[MAXDATASIZE]; /*./client 127.0.0.1 hello */
   char snd_buf[MAXDATASIZE];
   struct hostent *host;             /* struct hostent
                                      * {
                                      * char *h_name; // general hostname
                                      * char **h_aliases; // hostname's alias
                                      * int h_addrtype; // AF_INET
                                      * int h_length;
                                      * char **h_addr_list;
                                      * };
                                      */
   struct sockaddr_in server_addr;

   /* */
   fd_set readset, writeset;
   int check_timeval = ;
   struct timeval timeout={check_timeval,}; //阻塞式select, 等待1秒，1秒轮询
   int maxfd;
   int fp;
   int cir_count = ;
   int ret;

   if (argc < )
   {
     printf("Usage:%s [ip address] [any string]\n", argv[]);
     return ;
   }

   *snd_buf = '\0';
   strcat(snd_buf, argv[]);

   if ((fp = open(TFILE,O_WRONLY)) < )    //不是用fopen
   {
     perror("fopen:");
     exit();
   }

   if ((sockfd = socket(AF_INET, SOCK_STREAM, )) == -)
   {
     perror("socket:");
     exit();
   }

   server_addr.sin_family = AF_INET;
   server_addr.sin_port = htons(SERVPORT);
   inet_pton(AF_INET, argv[], &server_addr.sin_addr);
   memset(&(server_addr.sin_zero), , );

   /* create the connection by socket
    * means that connect "sockfd" to "server_addr"
    */
   if (connect(sockfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -)
   {
     perror("connect");
     exit();
   }

   /**/
   if (send(sockfd, snd_buf, sizeof(snd_buf), ) == -)
   {
     perror("send:");
     exit();
   }
   printf("send:%s\n", snd_buf);

   while ()
   {
     FD_ZERO(&readset);            //每次循环都要清空集合，否则不能检测描述符变化
     FD_SET(sockfd, &readset);     //添加描述符
     FD_ZERO(&writeset);
     FD_SET(fp,     &writeset);

     maxfd = sockfd > fp ? (sockfd+) : (fp+);    //描述符最大值加1

     ret = select(maxfd, &readset, NULL, NULL, &timeout);   // 非阻塞模式
     switch( ret)
     {
       case -:
         exit(-);
         break;
       case :
         break;
       default:
         if (FD_ISSET(sockfd, &readset))  //测试sock是否可读，即是否网络上有数据
         {
           recvbytes = recv(sockfd, rcv_buf, MAXDATASIZE, MSG_DONTWAIT);
           rcv_buf[recvbytes] = '\0';
           printf("recv:%s\n", rcv_buf);

           if (FD_ISSET(fp, &writeset))
           {
             write(fp, rcv_buf, strlen(rcv_buf));   // 不是用fwrite
           }
           goto end;
         }
     }
     timeout.tv_sec = check_timeval;    // 必须重新设置，因为超时时间到后会将其置零

     cir_count++;
     printf("CNT : %d \n",cir_count);
   }

 end:
   close(fp);
   close(sockfd);

   return ;
 }

 五、server端程序:
 /*
  * \brief
  * tcp server
  */
 #include <stdio.h>
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <string.h>
 #include <stdlib.h>
 #define SERVPORT 8080
 #define BACKLOG 10 // max numbef of client connection
 #define MAXDATASIZE 100

 int main(char argc, char *argv[])
 {
   int sockfd, client_fd, addr_size, recvbytes;
   char rcv_buf[MAXDATASIZE], snd_buf[MAXDATASIZE];
   char* val;
   struct sockaddr_in server_addr;
   struct sockaddr_in client_addr;
   int bReuseaddr = ;

   char IPdotdec[];

   /* create a new socket and regiter it to os .
    * SOCK_STREAM means that supply tcp service,
    * and must connect() before data transfort.
    */
   if ((sockfd = socket(AF_INET, SOCK_STREAM, )) == -)
   {
     perror("socket:");
     exit();
   }

   /* setting server's socket */
   server_addr.sin_family = AF_INET;         // IPv4 network protocol
   server_addr.sin_port = htons(SERVPORT);
   server_addr.sin_addr.s_addr = INADDR_ANY; // auto IP detect
   memset(&(server_addr.sin_zero),, );

   setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&bReuseaddr, sizeof(int));
   if (bind(sockfd, (struct sockaddr*)&server_addr, sizeof(struct sockaddr))== -)
   {
     perror("bind:");
     exit();
   }

   /*
    * watting for connection ,
    * and server permit to recive the requestion from sockfd
    */
   if (listen(sockfd, BACKLOG) == -) // BACKLOG assign thd max number of connection
   {
     perror("listen:");
     exit();
   }                                                                          

   while()
   {
     addr_size = sizeof(struct sockaddr_in);                                  

     /*
      * accept the sockfd's connection,
      * return an new socket and assign far host to client_addr
      */
     printf("watting for connect...\n");
     if ((client_fd = accept(sockfd, (struct sockaddr *)&client_addr, &addr_size)) == -)
     {
       /* Nonblocking mode */
       perror("accept:");
       continue;
     }                                                                        

     /* network-digital to ip address */
     inet_ntop(AF_INET, (void*)&client_addr, IPdotdec, );
     printf("connetion from:%d : %s\n",client_addr.sin_addr, IPdotdec);       

     //if (!fork())
     {
       /* child process handle with the client connection */                  

       /* recive the client's data by client_fd */
       if ((recvbytes = recv(client_fd, rcv_buf, MAXDATASIZE, )) == -)
       {
         perror("recv:");
         exit();
       }
       rcv_buf[recvbytes]='\0';
       printf("recv:%s\n", rcv_buf);                                          

       *snd_buf='\0';
       strcat(snd_buf, "welcome");                                            

       sleep();
       /* send the message to far-hosts by client_fd */
       if (send(client_fd, snd_buf, strlen(snd_buf), ) == -)
       {
         perror("send:");
         exit();
       }
       printf("send:%s\n", snd_buf);                                          

       close(client_fd);
       //exit(1);
     }                                                                        

     //close(client_fd);
   }

   return ;
 }